c# - 响应式扩展清理

Question

如果您使用 rx 进行长链调用，例如:

var responses = collectionOfHttpRequests.ToObservable()
.FromAsyncPattern(req.BeginGetResponse, req.EndGetResponse)
.Select(res => res.GetResponseBodyString()) // Extension method to get the body of the request
.Subscribe();

然后在操作完成之前调用处置，http 请求是否会被取消、关闭和正确处置，还是我必须以某种方式从方法链中选择 httprequests 并单独处置它们？

我遇到了一个问题，一个人可以同时发出多个 http 请求，我需要能够取消(而不是忽略)其中的一些/全部请求以节省网络流量。

Answer 1

当序列完成、错误或订阅被处置时，Rx 运算符链将自行清理。但是，每个运算符(operator)只会清理他们希望清理的内容。例如，FromEvent 将取消订阅该事件。

在您的情况下，Begin/End asynchronous pattern 不支持取消，所以 Rx 没有什么可以取消的。但是，您可以使用 Finally 调用 HttpWebRequest.Abort .

var observableRequests = collectionOfHttpRequests.ToObservable();

var responses = observableRequests
    .SelectMany(req => 
        Observable.FromAsyncPattern(req.BeginGetResponse, req.EndGetResponse)()
    )
    .Select(resp => resp.GetResponseBodyString())
    .Finally(() =>
    {
        observableRequests
            .Subscribe(req => req.Abort());
    })
    .Subscribe();