使用 Rx 下载文件（响应式编程）

Question

目前我正在使用此代码来获取文件：

WebClient webClient = new WebClient();
webClient.DownloadProgressChanged += webClient_DownloadProgressChanged;
webClient.DownloadFileCompleted += new AsyncCompletedEventHandler(webClient_ProgressCompleted);

我一直在查看文档以弄清楚如何使用 Rx 生成此代码。

我已经开始使用 FromEvent 创建一个 Observable 以监视 DownloadFileCompleted 事件。

Observable.FromEvent<?>(?, ?)

不过，我不知道如何填写这些 ?。

有人可以给我举个例子吗？

到目前为止，我尝试过：

Observable.FromEvent<AsyncCompletedEventArgs>(?1:addHandler, ?2:removeHandler).

然而，.net 要求我 ?1:addHandler 和 ?2:removeHandler 是 Action<Action<AsyncCompletedEventArgs>>（那是什么）？

Answer 1

这些重载非常棘手，我不得不 look them up every time。我提供了一些示例订阅代码来帮助您入门：

WebClient webClient = new WebClient();
var progressChangedObservable = Observable.FromEventPattern<DownloadProgressChangedEventHandler, DownloadProgressChangedEventArgs>(
    h => webClient.DownloadProgressChanged += h,
    h => webClient.DownloadProgressChanged -= h
);

var downloadCompletedObservable = Observable.FromEventPattern<AsyncCompletedEventHandler, AsyncCompletedEventArgs>(
    h => webClient.DownloadFileCompleted += h,
    h => webClient.DownloadFileCompleted -= h
);

progressChangedObservable
    .Select(ep => ep.EventArgs)
    .Subscribe(dpcea =>  Console.WriteLine($"{dpcea.ProgressPercentage}% complete. {dpcea.BytesReceived} bytes received. {dpcea.TotalBytesToReceive} bytes to receive."));

downloadCompletedObservable
    .Select(ep => ep.EventArgs)
    .Subscribe(_ => Console.WriteLine("Download file complete."));

var dummyDownloadPath = @"C:\temp\temp.txt";
webClient.DownloadFileAsync(new Uri(@"http://google.com"), dummyDownloadPath);

编辑：

根据@Enigmativity 的建议，可以以函数式风格执行所有这些代码，它还负责清理所有 IDisposable。但是，我不觉得它可读，因此不推荐它：

    Observable.Using(() => 
    {
        var webClient = new WebClient();
        webClient.Headers.Add("User-Agent: Other");
        return webClient;
    }, webClient =>
    Observable.Using(() => 
        Observable.FromEventPattern<DownloadProgressChangedEventHandler, DownloadProgressChangedEventArgs>(
                h => webClient.DownloadProgressChanged += h,
                h => webClient.DownloadProgressChanged -= h
            )
            .Select(ep => ep.EventArgs)
            .Subscribe(dpcea => Console.WriteLine($"{dpcea.ProgressPercentage}% complete. {dpcea.BytesReceived} bytes received. {dpcea.TotalBytesToReceive} bytes to receive.")),
        sub1 => Observable.Using(() =>
            Observable.FromEventPattern<AsyncCompletedEventHandler, AsyncCompletedEventArgs>(
                    h => webClient.DownloadFileCompleted += h,
                    h => webClient.DownloadFileCompleted -= h
                )
                .Select(ep => ep.EventArgs)
                .Subscribe(_ => Console.WriteLine("Download file complete.")),
            sub2 => webClient.DownloadFileTaskAsync(new Uri(@"http://google.com"), @"C:\temp\temp.txt").ToObservable()
        )
    )
)
    .Subscribe(_ => {} ); //Subscription required to trigger nested observables

使用 Rx 下载文件（响应式编程）

Download file using Rx (reactive programming)

.net

c#

system.reactive