Reactive Framework on Windows Phone 7

I didn’t have the opportunity to work with the Reactive framework from Microsoft until today when I saw the sample on how to emulate Location with it and this post in the developers forum. I can’t describe Reactive framework (Rx) here, but reading this series of posts by Matthew Podwysocki is a good start (see you in a few hours). To summarize, it transforms an Enumerable in a source of data (Observable) and allows you to react (Observer) to new items added, errors and completed operations. However an Enumerable isn’t necessarily finite: you can see a source (Mouse) of events (MouseClick) as a collection of items (the events) without end.

The Rx library is very handy when it comes to handle communication in an asynchronous world and Silverlight is such a world. The Observable implementation also exposes several Linq operators as Until, Join, Skip etc to manipulate the stream of events. The framework exists for Windows Phone 7 in the Microsoft.Phone.Reactive and System.Observable assemblies.

Here’s an example of the library with a WCF service:

var searchService = new MyServiceClient();
var observable = Observable.FromEvent<MyOperationCompletedEventArgs>(
	ev => searchService.MyOperationCompleted += ev,
	ev => searchService.MyOperationCompleted -= ev);

observable
	.ObserveOnDispatcher()
	.Subscribe(OnSearchCompleted);

void OnSearchCompleted(IEvent<MyOperationCompletedEventArgs> @event)
{
(...)
}

The method OnSearchCompleted is called on the UI thread so you can update widgets. Nothing special here, but how about  launching the search when the user enters text in a textbox after some time of inactivity ? Haha !

We can use the Throttle operator to handle this requirement:

var serviceObservable = Observable.FromEvent<MyOperationCompletedEventArgs>(
                ev => searchService.MyOperationCompleted += ev,
                ev => searchService.MyOperationCompleted -= ev);

var observableTextChanged = Observable
			.FromEvent<TextChangedEventArgs>(termTxtBx, "TextChanged")
.Throttle(TimeSpan.FromSeconds(1));

observableTextChanged.ObserveOnDispatcher().Subscribe(textArgs =>
									{
										IsProgressBarVisible = true;
										searchService.FindAsync(null, termTxtBx.Text, 0, 10);
									});

serviceObservable.Subscribe(OnSearchCompleted);