Lets imagine a common scenario. The user is typing in a text box trying to search for something in your app. We want to show results as the user types, however, we want to do so without starting a search on every single keystroke. So, we want to wait a little bit after a keystroke, and if no more keystrokes are launched within a small period then we initiate the search. However, if the user starts typing again we want to cancel the previous search.

Describing the scenario is quite simple, but the implemention can quite easily become very messy using conventional approaches.

Lets look at how we can implement the basic searching in response to key presses using Reactive Extensions (RX):

queryTextChangedObservable = 
Observable.FromEventPattern<TextChangedEventHandler, TextChangedEventArgs>
   (s => QTextBox.TextChanged += s, s => QTextBox.TextChanged -= s);
 
queryTextChangedObservable
.ObserveOn(SynchronizationContext.Current)
.Subscribe(async s =>
            {
                var textBox = (TextBox) s.Sender;
                await ViewModel.InitiateSearch(textBox.Text);
            });

We quickly realize that a lot of searches will be initiated. Another problem also arise. Normally RX ensures that calls to a subscription happen in sequential chronological order. However, since we are using an async lamda the  results of our remote request may intertwine and thus no longer be sequential.

Lets start by introducing a throttle that will constrain us from starting search until there is a pause in the key strokes:

queryTextChangedObservable
.Throttle(TimeSpan.FromMilliseconds(350))
.ObserveOn(SynchronizationContext.Current)
.Subscribe(...)

This will initiate a search every time the user stops typing for 350 milliseconds. This solves our first issues. The searches can still overlap though.

A simple way to cancel the previously initiated search would be to introduce a CancellationTokenSource on class level, and make sure we invoke a cancel on it everytime we start a new request:

private CancellationTokenSource cancellationTokenSource;

queryTextChangedObservable
.Throttle(TimeSpan.FromMilliseconds(350))
.Do(s => {if (cancellationTokenSource !=null) cancellationTokenSource.Cancel();})
.ObserveOn(SynchronizationContext.Current)
.Subscribe(async s =>
{
var textBox = (TextBox) s.Sender;
cancellationTokenSource = new CancellationTokenSource();

await ViewModel.InitiateSearch(textBox.Text, cancellationTokenSource.Token);
});

In the above code we use the “side-effect” method Do to explicitly show that we have some nasty side effects. Introducing the class level cancellation token source gets the job done, but it’s not very pretty. We much prefer to model the workflow with as little outside interaction as possible. So how can we model the above without the introduction of a class level variable?

To do this we need to be able to keep track of the previous elements CancellationTokenSource. Scan is nifty method that allows us to accumulate values between calls. We can use this to get access to the previous result and it’s cancellation token source.

<p style="margin-top: 0pt; margin-bottom: 16pt; line-height: 16pt; font-family: Calibri; font-size: 9.75pt; color: #444444;"> queryTextChangedObservable
.Throttle(TimeSpan.FromMilliseconds(350))
.Scan(new {cts = new CancellationTokenSource(), e = default(EventPattern<TextChangedEventArgs>)},
(previous, newObj) => { previous.cts.Cancel();
return new {cts = new CancellationTokenSource(), e = newObj};
})
.ObserveOn(SynchronizationContext.Current)
.Subscribe(async s =>
{
var textBox = (TextBox)s.e.Sender;
await ViewModel.InitiateSearch(textBox.Text, s.cts.Token);
});</p>
<p style="margin-top: 0pt; margin-bottom: 16pt; line-height: 16pt; font-family: Calibri; font-size: 9.75pt; color: #444444;">

This is an immense improvement since we no longer need to introduce a class level variable. The above 4 chained calls succesfully allowed us to throttle the initiated search calls while allowing us to cancel the previous calls. Using async in the subscribe introduce a little gotcha with regard to the sequential ordering, but nothing we couldn’t handle.

To reflect a little about how nice this solution is think about how you would implement this using a conventional approach. You would have to introduce a timer which got reset on every key stroke, then in the timer tick you could initiate searches, which you would then need to cancel in another method. All in all it would be extremely messy.