Reactive Extensions for .NET (Rx)

I’m pleased to announce a preview of the Reactive Extensions for .NET (Rx) on MSDN DevLabs.

Using Rx, programmers can write succinct declarative code to orchestrate and coordinate asynchronous and event-based programs based on familiar .NET idioms and patterns. Rx has a strong theoretical basis by using the duality between the classic Iterator and Observer design patterns to simplify controlling asynchrony. By combining the expressiveness of LINQ with the elegance of category theory, Rx allows programmers to write asynchronous code without performing cruel and unnatural acts.

As the name implies, reactive programs react to changes in their environment. Traditionally, programmers use locks and event handlers to coordinate these changes. Rx models asynchronous computations and events as push-based, or observable, collections, thus expanding the scope of the standard LINQ sequence operators and extension methods beyond the familiar pull-based, or enumerable, collections into the realm of reactive programming.

Practical and Pragmatic

The Reactive Extensions for .NET is a set of extension methods and an implementation of the LINQ standard sequence operators for the new IObservable<out T> and IObserver<in T> interfaces in .NET 4 and Silverlight 4. The observable interfaces were added to .NET to provide a common interface for push-based notifications that other .NET features and libraries can build on top of. Rx focuses on coordination and orchestration of event-based and asynchronous computations and leverages the new Task Parallel library as its underlying concurrency mechanism.

The Reactive Extensions can be used from any .NET language. In F#, .NET events are first-class values that implement the IObservable<out T> interface. In addition, F# provides a basic set of functions for composing observable collections and F# developers can leverage Rx to get a richer set of operators for composing events and other observable collections.

The Silverlight Toolkit uses Rx to power the Silverlight Toolkit Drag and Drop Framework, a subset of the WPF APIs for initiating and interacting with drag operations. The Drag and Drop Framework adds drag and drop support to core controls such as Treeview, DataGrid, ListBox. Rx drastically reduced development time by allowing drag operations, which are sequences of user events, to be described declaratively. Silverlight creates visual elements asynchronously and as a result it is often necessary to write asynchronous test code to ensure that a component has been correctly created. The addition of Rx to the Silverlight unit testing framework allows developers to write reliable event-based tests without sacrificing readability.

Show me Some Code

The example below shows a simple AJAX-style reactive program that translates English using the Bing translation service and displays the results.

The input field in the page is exposed as an observable collection of strings that produces a value whenever the user has paused typing for a half second. The GetKeyUpEvents extension method exposes the standard .NET KeyUp event as an observable collection.

    IObservable<string> words = (from keyup in Input.GetKeyUpEvents() select Input.Value).Throttle(TimeSpan.FromSeconds(.5))

To access the Bing translation service, we simply convert a simple WCF service reference that uses the standard .NET BeginInvoke/EndInvoke or event-based async pattern into a function that returns a singleton observable collection via one of the standard helper functions that Rx provides.

    IObservable<TranslationResponse> Translate(this string text, string sourceLanguage, string destinationLanguage){…}

The coordination between the various computations uses a LINQ query that sends requests to the Bing translation service to translate each word into Dutch, French, and Spanish and then waits for the first two of three results to return using a join pattern.

    var translations = from word in words

            let dutch = Bing.Translate(word, "en", "nl")

            let french = Bing.Translate(word, "en", "fr")

            let spanish = Bing.Translate(word, "en", "es")

            from results in Observable.Join

                 ( dutch.And(spanish).Then((d, s) =>
                   new { Dutch = d, French = Bing.NoResult, Spanish = s })

                 , dutch.And(french).Then((d, f) =>
                   new { Dutch = d, French = f, Spanish = Bing.NoResult })

                 , french.And(spanish).Then((f, s) =>
                   new { Dutch = Bing.NoResult, French = f, Spanish = s })

                 ).Until(words)

             select results;

Finally, we subscribe to the results of the translations and update the UI once we get notified of a change.

    translations.Subscribe(result =>

    {

        Dutch.InnerText = result.Dutch.GetTranslatedTerm();

        French.InnerText = result.French.GetTranslatedTerm();

        Spanish.InnerText = result.Spanish.GetTranslatedTerm();

    });

Developers don’t need to worry about executing the subscription on the correct thread; the Rx runtime takes care of it.

Bonus Material

Besides the extension methods on observable collections, Rx also contains a number of more experimental types and namespaces that implement some of the new extension methods for observable collections over enumerable collections and an experimental back port of PLINQ to .NET 3.5 that are available to developers for experimentation.

Let’s Play

You can download Rx from DevLabs. Tell us what you think on the project forum, and check out videos and screencasts on Channel 9.

Namaste!