Bing it on, Reactive Extensions! – Story, code and slides
5.00/5 (3 votes)
A sample of using Reactive Extensions.
I held a presentation at UBISOFT Buchares headquarters for the RONUA local programmers user group recently as I’ve announced earlier. Here’s the contents, step-by-step, final code and slides.
————- [scroll way down for the downloads]
I was recently tasked with rewriting an app component by leveraging Reactive Extensions. I knew little about Rx (the short form of Reactive Extensions) and all I remembered was that it has two interfaces IObservable and IObserver and it seemed dull at that time.
Basically the component enables search without needing to hit ENTER or a “Go!” button, although provides for these. After the user finishes typing an async request goes to the data store and searches for the phrase entered and fetches the results. In the original implementation the component used a lot of timers, event handlers, private fields all making up a nice spaghetti bowl of code.
Let’s do this step by step and see how our little (demo) app develops. Fire up Visual Studio 2012 and start a new WPF project (.NET 4.5 preferably). The very next thing we’ll install Rx. Right click on the project in the Solution Explorer and select “Manage NuGet Packages” (you can also use the Package Manager Console if you like it better). Search online for “Reactive Extensions”.
In the result lists (this requires a functional internet connection) select ‘Reactive Extensions – WPF Helpers’ (the nice thing about NuGet packages is that it automatically resolves and installs all the dependencies). Accept the EULAs (you know what’s the most common lie told these days? “I have read and accepted the terms of the license” 
).
In our demo we will use Bing as the data store which we’ll target through our searches (sorry, Google was too difficult to setup, offered less search requests per month and no C# demo code. Thanks Google, thanks again.). In order to do this you will need a Microsoft Account (I guess we all have one these days). Go to http://www.bing.com/developers/ and then select “Search API” -> Start now (this will lead you to https://datamarket.azure.com/dataset/5BA839F1-12CE-4CCE-BF57-A49D98D29A44 ). There are paid subscriptions and a free subscription. Hit signup and go through the process (leave a comment if you are unable to go through this process).
In the end you will need to obtain the (Primary) Account Key and the Customer ID. These are available under “My Account” -> Account Information ( https://datamarket.azure.com/account ). We’ll use these later so save them. Also, don’t share them with other people because these are your credentials. Also visit “My Data” ( https://datamarket.azure.com/account/datasets ) and click on “Bing Search API”‘s “Use” link (far right, https://datamarket.azure.com/dataset/explore/bing/search ). Capture the “URL for current expressed query” : “https://api.datamarket.azure.com/Bing/Search/v1/Composite”. We’ll also need these later.
Finally just below the subscription list there is a “.NET C# Class library” link (https://datamarket.azure.com/dataset/explore/getproxy/5ba839f1-12ce-4cce-bf57-a49d98d29a44) which will get you a “BingSearchContainer.cs” file. Download this file into the project’s folder and include it in your project. At this point the project will not compile as it is missing a reference to System.Data.Services.Client. Add this reference and compile the project.
Now let’s build the UI. I must say that for the sake of the demo we will not implement this little app by abiding to all good practices and patterns, such as MVVM. If you do end up implementing this for production purposes please take care to abide to all good practices. Getting back to our development, let’s open MainWindow.xaml and prepare a simple layout with a thin row above, on which we’ll place a textbox, two buttons and a “busy indicator”. On the second row (the most of the app space, we’ll place a listbox with results :
<Window x:Class="BingItOn.MainWindow"
xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
xmlns:bing="clr-namespace:Bing"
mc:Ignorable="d"
Title="Bing it on!"
Width="800"
Height="600"
FocusManager.FocusedElement="{Binding ElementName=TxtSearch}">
<Grid>
<Grid.RowDefinitions>
<RowDefinition Height="Auto"/>
<RowDefinition />
</Grid.RowDefinitions>
<Grid.ColumnDefinitions>
<ColumnDefinition />
<ColumnDefinition Width="Auto"/>
</Grid.ColumnDefinitions>
<TextBox x:Name="TxtSearch" />
<StackPanel Orientation="Horizontal" Grid.Column="1">
<Button x:Name="BtnForceSearch" IsDefault="True">Go!</Button>
<Button x:Name="BtnStopSearch" IsCancel="True">Stop</Button>
<Grid x:Name="GrdBusy" Background="Red" Visibility="Hidden"
Width="20" />
</StackPanel>
<ListBox x:Name="LstResults" Grid.Row="1" Grid.ColumnSpan="2">
<ListBox.ItemTemplate>
<DataTemplate>
<StackPanel
d:DataContext="{d:DesignInstance {x:Type bing:WebResult}}">
<TextBlock Text="{Binding Title}"></TextBlock>
<TextBlock Text="{Binding Url}"></TextBlock>
<TextBlock>
<Hyperlink NavigateUri="{Binding Url}"
RequestNavigate="OnUrlNavigate">
<TextBlock Text="{Binding DisplayUrl}"/>
</Hyperlink>
</TextBlock>
</StackPanel>
</DataTemplate>
</ListBox.ItemTemplate>
</ListBox>
</Grid>
</Window>
The Grid “GrdBusy” is set to Hidden and not “Collapsed” in order to avoid having the buttons shift left-right. It also requires a width in order to be drawn on the screen. The weird d:DataContext is an attached property that lets the IDE (ReSharper uses this) know what kind of objects will be fed into the “LstResults” listbox.
Now let’s go in the code behind (this sounds anti-pattern but remember it’s just a quick demo – otherwise we would have had a ViewModel and so on) and declare two private constants with the account key and customer id (we’ll need these to do the actual search).
Let’s also build a small method that takes in a string and returns an IEnumerable<webresult> (WebResult is a class in the Bing namespace), let’s call it SearchBing. At this point the MainWindow class should look like :
using System;
using System.Collections.Generic;
using System.Linq;
using Bing;
namespace BingItOn
{
public partial class MainWindow
{
private const string AccountKey = @"<YOUR_ACTUAL_ACCOUNT_KEY>";
private const string CustomerId = @"<YOUR_ACTUAL_CUSTOMER_ID>";
public MainWindow()
{
InitializeComponent();
}
private IEnumerable<WebResult> SearchBing(string searchPhrase)
{
var uri = new Uri(@"https://api.datamarket.azure.com/Bing/Search/v1/Web");
var container = new BingSearchContainer(uri);
container.Credentials = new NetworkCredential(CustomerId, AccountKey);
var query = container.Web(searchPhrase, null, null, null, null, null, null, null);
return query.Take(10).ToArray();
}
private void OnUrlNavigate(object sender, RequestNavigateEventArgs e)
{
Process.Start(e.Uri.ToString());
}
}
}
Step 1 – let’s have the app search async and display results as we type.
Let’s create a “SetupSubscriptions” void instance method that we’ll call from the constructor after InitializeComponent();. In this method we’ll create a new ‘Observable’. That is an instance of a class that implements IObservable<t>. This thin interface has one method : Subscribe. This enables us to subscribe with a method which will be called back (callback) whenever the sequence, because a sequence of data is what an IObservable models, produces a new item. In case of an event observable the item will be an encapsulation of the event sender and the event arguments ('object sender, EventArgs e' ring any bells?). We’ll select the new text and then we’ll subscribe to this observable like so :
public MainWindow()
{
InitializeComponent();
SetupSubscriptions();
}
private void SetupSubscriptions()
{
var textChangedObservable =
Observable.FromEventPattern(TxtSearch, "TextChanged");
var textObservable =
from evt in textChangedObservable select ((TextBox)evt.Sender).Text;
textObservable.Subscribe(OnTextChanged);
}
private void OnTextChanged(string text)
{
var data = SearchBing(text);
LstResults.Items.Clear();
foreach (var result in data)
{
LstResults.Items.Add(result);
}
}
Run the app (if you have aggressive firewall software you’ll be prompted to accept this little app’s request to connect externally) and notice how things happen synchronously, sluggishly. Type a character, wait for the Bing results, type another character and so on. This is not so nice. Let’s make it async. 
We do this by creating an observable of async web results and we’ll combine the text observable (with a ‘inner join’) with this new observable. We’ll also change the OnTextChanged method to OnResultsArrived and its contents. Sounds complicated but it’s not really. Let’s see how :
private void SetupSubscriptions()
{
var textChangedObservable = Observable.FromEventPattern(TxtSearch, "TextChanged");
var textObservable = from evt in textChangedObservable select ((TextBox)evt.Sender).Text;
var searchBingFunc = new Func<string, IEnumerable<WebResult>>(SearchBing);
var searchBingObservable = Observable.FromAsyncPattern<string, IEnumerable<WebResult>>(
searchBingFunc.BeginInvoke, searchBingFunc.EndInvoke);
var combinedObservable = from text in textObservable
from result in searchBingObservable(text)
select result;
combinedObservable.Subscribe(OnResultsArrived);
}
private void OnResultsArrived(IEnumerable<WebResult> results)
{
LstResults.Items.Clear();
foreach (var result in results)
{
LstResults.Items.Add(result);
}
}
Upon the first run we get hit with an InvalidOperationException at the first line of code of the OnResultsArrived because we’re not on the UI thread but on a worker thread. Resist the urge to Dispatcher.Invoke (or BeginInvoke) into this. We’ll change the Subscription code from combinedObservable.Subscribe(OnResultsArrived) to combinedObservable.ObserveOn(this).Subscribe(OnResultsArrived).
Run the app again. Now things work a bit better but still sluggish (the results presentation is the culprit) and more than this, a search request is made for each keystroke (or at least any command that changes the text). We can easily prevent this with a little… Reactive Extension! Called Throttle.
step 2 – let’s start the search only if the user stopped typing
We’ll apply Throttle on the textObservable in the combinedObservable definition in SetupSubscriptions like so :
var combinedObservable = from text in textObservable.Throttle(TimeSpan.FromSeconds(0.5))
from result in searchBingObservable(text)
select result;
Running the app again shows much better performance and less requests to the server(s). What Throttle essentially does is waiting for the sequence to calm down and if after 0.5 seconds no more changes have been made, it stops filtering and allows the item to pass out of the sequence. Neat, ha?
We have one more nice extension that we can chain after Throttle and that’s DistinctUntilChanged. This allows to pass only distinct results. So if we type “test” in the textbox and then quickly (in less than those 500 milliseconds) type another “t” and then quicly delete it there won’t be requests made for “testt” and then another one for “test” again.
var combinedObservable = from text in textObservable
.Throttle(TimeSpan.FromSeconds(0.5))
.DistinctUntilChanged()
from result in searchBingObservable(text)
select result;
Step 3 – let’s drop an ongoing search if in the meantime we need to start another one
Now we’re finally seeing the beauty of reactive extensions! By simply combining observable (either chaining them or joining them) and/or applying filters and modifiers we’re obtaining a lot of functionality with little code. It’s more elegant, easier to understand (after the learning curve albeit), easier to test and less prone to bugs since the functionality is outsourced to the Rx libs which are carefully written by the guys at Microsoft and thoroughly tested.
searchBingObservable(text) in the combined observable definition is a single-value observable that ‘spits’ out a single result at most once in its life, that is an “instance” of IEnumerable<webresult>. (I used the quotes because a closer to the truth expression would have been an “instance of a class that implements IEnumerable<webresult>, array of WebResults to be more precise”).
The .TakeUntil(anotherSequence) extension method found in Rx filters a sequence in a way that it lets pass everything until its parameter (‘anotherSequence’) produces an item. This way we’d be ignoring the results when they’d came (effectively cancelling the request on the server(s) is not possible in a HTTP(S) system). We’ll just add TakeUntil in the mix like so:
private void SetupSubscriptions()
{
var textChangedObservable = Observable.FromEventPattern(TxtSearch, "TextChanged");
var textObservable = from evt in textChangedObservable select ((TextBox)evt.Sender).Text;
var searchBingFunc = new Func<string, IEnumerable<WebResult>>(SearchBing);
var searchBingObservable = Observable.FromAsyncPattern<string, IEnumerable<WebResult>>(
searchBingFunc.BeginInvoke, searchBingFunc.EndInvoke);
var tamedTextObservable = textObservable.Throttle(TimeSpan.FromSeconds(0.5)).DistinctUntilChanged();
var combinedObservable = from text in tamedTextObservable
from result in searchBingObservable(text).TakeUntil(tamedTextObservable)
select result;
combinedObservable.ObserveOn(this).Subscribe(OnResultsArrived);
}
Everything’s fine until someone trips over the wireless cable (kidding, the net goes down whatsoever). In that case the exception will take down the whole app although it shouldn’t.
step 4 – Exception handling
When a sequence (observable) encounters an error it calls OnError on its subscriber(s) and then it terminates, i.e. it won’t produce anything anymore. It’s faulted. Therefore in such a case we’ll choose to present the error and redo the subscriptions :
//...
combinedObservable.ObserveOn(this).Subscribe(OnResultsArrived, OnError);
}
private void OnError(Exception obj)
{
MessageBox.Show("An error has occured." + Environment.NewLine + Environment.NewLine + obj);
SetupSubscriptions();
}
step 5- let’s implement the busy indicator
We’ll just define a helper method that toggles the busy indicator state and call it from SearchBing, OnResultsArrived, and OnError methods (at their start) accordingly (true for SearchBing and false for the others) :
private void SetBusyIndicatorState(bool state)
{
Dispatcher.BeginInvoke(new Action(() => {
GrdBusy.Visibility = state ? Visibility.Visible : Visibility.Hidden; }));
}
Let’s run. It works, not a state-of-the-art like an animated spinner or something but that would be beside the point of this demo anyway.
step 6 – let’s implement the Go! button (the force search button)
We’ll start by observing the button click event and then merging this observable with the tamed text input. We’ll also increase significantly the throttle time span in order to … observe (pun intended) the effect of hitting ENTER (or mouse-clicking the button) :
//...
var tamedTextObservable = textObservable.Throttle(TimeSpan.FromSeconds(3.5)).DistinctUntilChanged();
var forceSearchButtonObservable =
from text in Observable.FromEventPattern(BtnForceSearch, "Click") select TxtSearch.Text;
var combinedObservable =
from text in tamedTextObservable.Merge(forceSearchButtonObservable)
from result in searchBingObservable(text).TakeUntil(tamedTextObservable)
select result;
//...
Running the application, typing ‘test’ (for example) and quickly hitting ENTER we’ll see that the search commences instantly (or almost). However we’ll also observe that after we get the results another query is issued for the same text and, of course, has the same results.
The essence of the issue is that the expression “tamedTextObservable.Merge(forceSearchButtonObservable)” will produce an item (search text) at the ENTER keypress and another one later from the tamed observable. The quick fix is to add an “DistinctUntilChanged” at its end. This fixes it until…
The boss comes requesting that hitting ENTER even after the results have been loaded for the current expression will redo the search, just like a Refresh (F5) would do in a browser. Now we’re stuck we need to remove the “DistinctUntilChanged” but we’d then be back to the double search issue that we had later. It may seem we can’t outsmart this.
But after a few hours of head-banging it hit me : we’ll switch from string (the type produced by the sequences and accepted by the search method) to a custom type that we’ll call SearchRequest:
public class SearchRequest
{
public string Text { get; set; }
public bool Force { get; set; }
}
Instead of just producing the TxtSearch’s .Text value we’ll be producing instances of this SearchRequest class filling the TextBox’s Text property value in the Text property and setting Force to true only if produced by the force button click. We’ll also use this opportunity to treat the case when the textbox becomes empty.
private void SetupSubscriptions()
{
var textChangedObservable = Observable.FromEventPattern(TxtSearch, "TextChanged");
var textObservable =
from evt in textChangedObservable select new SearchRequest { Text = TxtSearch.Text };
var searchBingFunc = new Func<SearchRequest, IEnumerable<WebResult>>(SearchBing);
var searchBingObservable =
Observable.FromAsyncPattern<SearchRequest, IEnumerable<WebResult>>(
searchBingFunc.BeginInvoke, searchBingFunc.EndInvoke);
var tamedTextObservable = textObservable.Throttle(TimeSpan.FromSeconds(3.5)).DistinctUntilChanged();
var forceSearchButtonObservable =
from text in Observable.FromEventPattern(BtnForceSearch, "Click")
select new SearchRequest { Text = TxtSearch.Text, Force = true };
var combinedObservable =
from request in tamedTextObservable.Merge(forceSearchButtonObservable).DistinctUntilChanged()
from result in searchBingObservable(request).TakeUntil(tamedTextObservable)
select result;
combinedObservable.ObserveOn(this).Subscribe(OnResultsArrived, OnError);
}
//...
private IEnumerable<WebResult> SearchBing(SearchRequest searchRequest)
{
if (string.IsNullOrWhiteSpace(searchRequest.Text)) return Enumerable.Empty<WebResult>();
SetBusyIndicatorState(true);
var container = new BingSearchContainer(new Uri(Url))
{
Credentials = new NetworkCredential(CustomerId, AccountKey)
};
var query = container.Web(searchRequest.Text, null, null, null, null, null, null, null);
return query.Take(10).ToArray();
}
We’ll run the app and see that not much has changed. That’s because now the DistinctUntilChanged at the end of tamedTextObservable definition is using the Equals of the SearchRequest class and the default implementation of a class is to compare the reference values and not the semantic value. We’ll also override GetHashCode since this is a good practice that can defend us from weird bugs. Let’s implement this cleverly placing priority for the case when the request is forced :
internal class SearchRequest
{
public string Text { get; set; }
public bool Force { get; set; }
public override bool Equals(object obj)
{
var older = this;
var newer = (SearchRequest)obj;
if (newer == null || newer.Text != older.Text) return false;
return !newer.Force;
}
public override int GetHashCode()
{
return (Text ?? string.Empty).GetHashCode() ^ Force.GetHashCode();
}
}
This reminds of a strange case when ‘this’ can be null in an instance context but I won’t treat it in the Equals overriding. Run the app. Problem fixed! This feels good.
Step 7 – let’s implement the stop button
This will need to do several things : clear the textbox, clear the results and stop any ongoing search if any. Although there probably are many ways to implement this at this stage of development I will choose, what I think to be, a simple one : I will handle the click event and clear the textbox and, separately, I will observe the click event and use it (indirectly) in the TakeUntil of the async observable to stop an ongoing search. I will also use this opportunity to set the throttle time at a more decent value, such as 0.7 seconds.
<Button x:Name="BtnStopSearch" IsCancel="True"
Click="OnStopRequest">Clear</Button>
private void SetupSubscriptions()
{
var textChangedObservable = Observable.FromEventPattern(TxtSearch, "TextChanged");
var textObservable =
from evt in textChangedObservable select new SearchRequest { Text = TxtSearch.Text };
var searchBingFunc =
new Func<SearchRequest, IEnumerable<WebResult>>(SearchBing);
var searchBingObservable =
Observable.FromAsyncPattern<SearchRequest, IEnumerable<WebResult>>(
searchBingFunc.BeginInvoke, searchBingFunc.EndInvoke);
var tamedTextObservable =
textObservable.Throttle(TimeSpan.FromSeconds(0.7)).DistinctUntilChanged();
var forceSearchButtonObservable =
from click in Observable.FromEventPattern(BtnForceSearch, "Click")
select new SearchRequest { Text = TxtSearch.Text, Force = true };
var stopButtonObservable =
from click in Observable.FromEventPattern(BtnStopSearch, "Click")
select new SearchRequest { Text = string.Empty };
var anyInputObservable =
tamedTextObservable.Merge(forceSearchButtonObservable).Merge(stopButtonObservable);
var combinedObservable =
from request in anyInputObservable.DistinctUntilChanged()
from result in searchBingObservable(request).TakeUntil(anyInputObservable)
select result;
combinedObservable.ObserveOn(this).Subscribe(OnResultsArrived, OnError);
}
//...
private void OnStopRequest(object sender, RoutedEventArgs e)
{
TxtSearch.Clear();
}
This app can be refined much more and finally ditched for a good ol’ browser 
)) but for now let’s call it a wrap!
Here’s the final code (VS 2012 / .NET 4.5 solution): BingItOn solution
and the slides from the presentation: Bing it on Reactive Extensions
Want to learn more? Try the best resource for Rx that I’ve used to develop this tiny intro : http://introtorx.com