Rx samples with WinRT by ThinqLinq

Recently I decided to rebuild one of my presentation computers using the Windows 8 Consumer Preview and Visual Studio 11 Beta. While it is working quite well for my typical usage, I did run into a bit of an issue when prepping my Rx talks for the Rocky Mountain Trifecta this past weekend.

In the past, when I'm giving my Practical Rx talk, I show Silverlight, WPF, WP7 and RxJs samples. Unfortunately, the Windows 8 Consumer Preview does not support the Windows Phone 7 emulator and Visual Studio 11 does not support the WP7 targets. As an alternative, i decided to port the WP7 demos over to a Metro style Win-RT application, which you can download now.

The main changes were due to slightly different event names, but other all of the Rx methods that I needed were available in the Rx Beta 2 experimental build that Bart introduced on the Rx Blog. However, when preping the prototypical Dictionary Suggest sample, I ran into a TargetInvocationException when trying to dispose the event handler. Searching a bit further into the call-stack, I found that the exception was caused when disposing the observable subscriptions due to the TakeUntil clause. Even though the Throttle is what pushes this code off of the dispatcher thread. For reference, here's the broken code method:

    var svc = new SimpleService.SimpleServicesvcClient();
    var results = new ObservableCollection();
    Translations.ItemsSource = results;

    IObservable inputStrings =
        from keyup in Observable.FromEventPattern
            (InputText, "TextChanged")
        select InputText.Text;

    var svcResults = 
        from text in inputStrings
        .Throttle(TimeSpan.FromMilliseconds(250))
        from result in svc.DoSomethingCoolAsync(new DoSomethingCoolRequest { input = text })
        .ToObservable()
        .TakeUntil(inputStrings)
        select String.Format("{0} - {1}", text, result.DoSomethingCoolResult);

    svcResults
        .ObserveOnDispatcher()
        .Subscribe(result => results.Insert(0, result));
         

Thanks to a bit of additional snooping by the Rx and WinRT teams, we found that the real reason here is that we are attaching the TextChanged event handler on the core dispatcher thread. However, when we use the Throttle, the operation is shifted to another thread. When we add the TakeUntil, we instruct Rx to dispose of the current subscription when the second stream starts. Of course, the new keypress also starts another stream of observables.  While this works without errors in the .Net world (Silverlight/WP7/WPF), WinRt is a bit pickier and requires that the remove handler be called on the same thread that was used when the handle was added. To fix this situation, we need to explicitly force the subscription onto the Dispatcher thread using SubscribeOnDispatcher. Here's the fix:

    var svc = new SimpleService.SimpleServicesvcClient();
    var results = new ObservableCollection();
    Translations.ItemsSource = results;

    IObservable inputStrings =
        from keyup in Observable.FromEventPattern
            (InputText, "TextChanged")
        .SubscribeOnDispatcher()
        select InputText.Text;

    var svcResults = 
        from text in inputStrings
        .Throttle(TimeSpan.FromMilliseconds(250))
        from result in svc.DoSomethingCoolAsync(new DoSomethingCoolRequest { input = text })
        .ToObservable()
        .TakeUntil(inputStrings)
        select String.Format("{0} - {1}", text, result.DoSomethingCoolResult);

    svcResults
        .ObserveOnDispatcher()
        .Subscribe(result => results.Insert(0, result));