Rx 2.1:如何在Dispatcher上正确订阅和观察

Question

首先,我使用最新的Rx,即2.1.据我所知,当Rx变为2时,很多事情都发生了变化,所以我真的很期待收到最新的答案.提前致谢.

我正在为Rx实现一个经典任务:观察TextBox的文本(确切地说是来自WPToolkit的AutoCompleteBox),以便向用户提供建议列表.建议来自网络,我想使用这些普通的Rx好东西,如Throttle,DistinctUntilChanged等.

我也在使用最近发布的适用于Windows Phone 8的便携式HttpClient,因为它提供了基于任务的异步API,这很不错.

我遇到的问题是在读取Text 'AutoCompleteBox` 的值时的跨线程访问.这是代码:

var http = new HttpClient();
var searchFunc = Observable.FromAsync<HttpResponseMessage>(() => 
            http.GetAsync(FormatUrl(SEARCH_URL, "DE", new GeoCoordinate(51, 13), searchBox.Text /* <-- causes exception */, 10, "")));

var uithread = new SynchronizationContextScheduler(SynchronizationContext.Current);
var textChange = Observable.FromEventPattern<RoutedEventArgs>(searchBox, "TextChanged")                             
        .Throttle(TimeSpan.FromMilliseconds(800))
        .DistinctUntilChanged()     
        .SubscribeOn(uithread)           
        .SelectMany(searchFunc)                
        .Select(async (resp) => SearchResultsParser.ParseSearchResults(await resp.Content.ReadAsStreamAsync(), new GeoCoordinate(51, 13)))
        .Select(async (results) => searchBox.ItemsSource = await results)
        .ObserveOn(uithread)
        .Subscribe();

searchFunc执行时会发生异常.我从VS看到它在Worker Thread上执行,尽管我使用了SubscribeOn.

这是使用的示例SynchronizationContextScheduler,但我也尝试过SubscribeOnDispatcher,结果相同.看起来我错过了一些重要的ObserveOn东西,或者可能是关于Observable.FromAsync.你能不能指出我的错误？

Answer 1

SubscribeOn是几乎从来没有你想要的东西-你可能会认为这是指"我的Subscribe方法运行",但它实际上是指"到实际的布线IDisposable(处置)运行" - ObserveOn是等效的"这是我希望我的实际Subscribe代码执行"

参考:Observable.SubscribeOn和Observable.ObserveOn