gistfile1.cs

IObservable<CacheEntry> downloadUriAndSave(Uri uri, string cacheDirectory, DateTimeOffset? expiration, Dictionary<string, string> headers, string content)
{
    var request = Observable.Defer(() =>
    {
        var hwr = WebRequest.Create(uri);
        if (headers != null)
        {
            headers.ForEach(x => hwr.Headers[x.Key] = x.Value);
        }

        if (content == null)
        {
            return Observable.FromAsyncPattern<WebResponse>(hwr.BeginGetResponse, hwr.EndGetResponse)();
        }

        var buf = Encoding.UTF8.GetBytes(content);
        return Observable.FromAsyncPattern<Stream>(hwr.BeginGetRequestStream, hwr.EndGetRequestStream)()
            .SelectMany(x => Observable.FromAsyncPattern<byte[], int, int>(x.BeginWrite, x.EndWrite)(buf, 0, buf.Length))
            .SelectMany(_ => Observable.FromAsyncPattern<WebResponse>(hwr.BeginGetResponse, hwr.EndGetResponse)());
    });

    var result = request.Timeout(timeout).Retry(retries);
    return result.SelectMany(x => processWebResponse(uri, headers, content, expiration, x));
}

IObservable<CacheEntry> processWebResponse(Uri uri, Dictionary<string, string> headers, string content, DateTimeOffset? expiration, WebResponse response)
{
    var key = getKeyForRequest(uri, headers, content);
    var entry = new CacheEntry() {Key = key, ExpirationDate = expiration};

    var existingEntry = requestCache.FirstOrDefault(x => x.Key == key);
    if (response.Headers["Cache-control"] == "no-store")
    {
        existingEntry = null;
    }

    if (response.Headers["ETag"] != null)
    {
        entry.ETag = response.Headers["ETag"];
        if (existingEntry != null && existingEntry.ETag == entry.ETag && entry.ETag != null)
        {
            return Observable.Return(existingEntry);
        }
    }

    int lastModified;
    if (Int32.TryParse(response.Headers["Last-Modified"] ?? "", out lastModified))
    {
        entry.LastModified = lastModified;
        if (existingEntry != null && entry.LastModified == existingEntry.LastModified)
        {
            return Observable.Return(existingEntry);
        }
    }

    int expires;
    if (Int32.TryParse(response.Headers["Expires"], out expires))
    {
        DateTimeOffset expireDate = new DateTimeOffset(70, 1, 1, 0, 0, 0, TimeSpan.Zero) + TimeSpan.FromSeconds(expires);
        entry.ExpirationDate =  (expireDate < (entry.ExpirationDate ?? DateTimeOffset.MaxValue)) ? expireDate : entry.ExpirationDate;
    }

    Stream cacheFile = null;
    try
    {
        cacheFile = File.OpenWrite(Path.Combine(cacheDirectory, key));
        response.GetResponseStream().CopyTo(cacheFile);
    }
    catch(Exception ex)
    {
        return Observable.Throw<CacheEntry>(ex);
    }
    finally
    {
        if (cacheFile != null)
        {
            cacheFile.Dispose();
        }
    }

    return Observable.Return(entry);
}