响应式扩展更新UI

Question

我多次发起异步网络请求（根据条件可能是两次、三次或甚至六次）

        var request = HttpWebRequest.CreateHttp(url);

        var observableRequest = Observable.FromAsyncPattern<WebResponse>(
            request.BeginGetResponse, request.EndGetResponse);

        Observable.Timeout(observableRequest.Invoke(), TimeSpan.FromSeconds(120)).
            Subscribe(response => { HandleListResult(response); },
            exception => { HandleListResultTimeOut(exception); });

我在ViewModel中有一个Collection（列表），它与一个LisBox绑定，我想在每个响应返回后继续添加到该集合中。

使用Reactive Extensions实现这一点的最佳实践是什么？如果有人能给我展示一些示例代码，那就太好了！

提前致谢

- user636525

---

由于您将通过引发PropertyChanged或ObservableCollection来更新UI，因此最好使用ObserveOnDispatcher。 - Maciek

Answer 1

你可以直接将URL流转换为流：

    public static IObservable<Stream> RequestToStream(
        this IObservable<string> source, TimeSpan timeout)
    {
        return
            from wc in source.Select(WebRequest.Create)
            from s in Observable
                .FromAsyncPattern<WebResponse>(wc.BeginGetResponse,
                    wc.EndGetResponse)()
                .Timeout(timeout, Observable.Empty<WebResponse>())
                .Catch(Observable.Empty<WebResponse>())
            select s.GetResponseStream();
    }

接下来，您需要观察UI上的响应，您需要使用.ObserveOnDispatcher()，例如：

        Observable
            .Return("www.msdn.com")
            .RequestToStream(TimeSpan.FromSeconds(1))
            .ObserveOnDispatcher()
            .Subscribe(request => UpdateUI(Request));

- Sergey Aldoukhov

---

谢谢！在更新绑定到 ListBox 的 List<string> 对象之前，我需要锁定它吗？因为同一个处理程序将被多次调用，我想知道当第二个处理程序尝试在第一个处理程序已经更新它时会发生什么。谢谢！ - user636525

不可能发生。每个更新UI的请求都会在同一个UI线程上发生 - 因此它们将被序列化。无需锁定。 - Sergey Aldoukhov

你也可以使用Defer和Retry来增加调用的可靠性（即如果其中一个超时，再尝试一次），但这里使用Rx非常棒！ - Ana Betts

Answer 2

在ReactiveUI中，这是通过CreateCollection()实现的。

IObservable<string> source; // Maybe this is a Subject<string> or whatever

myBoundCollection = source
    .SelectMany(webServiceCall) // This is your FromAsyncPattern func
    .CreateCollection();  // Pipe the Observable to a Collection

ReactiveUI处理了所有的ObserveOn操作，以确保它们同步并在正确的线程上运行。此调用立即返回一个空列表，然后随着结果的到来，集合会被填充。

- Ana Betts