I have a problem whereby I need to iterate over a potentially large number of rows in a UI grid control, the iteration needs to take place on the UI thread but the writing (which in this instance write the data to Excel) can occur on a background thread which should make the UI a little more responsive.

Now this might not be the best solution but it seems to work better than other more synchronous solutions. One thing to note is that this current design expects the call to the method to be on the UI thread and hence doesn’t marshal the call to the grid control onto the UI thread (it assumes it’s on it).

Within the UI iteration method I create a new observable using

var rowObservable = Observable.Create<string>(observer =>
{
   // iterate over grid records calling the OnNext method 
   // for example
   foreach(var cell in Cells)
   {
      observer.OnNext(cell.Value);
   }

   observer.OnCompleted();
   return Disposable.Empty;
});

In the above code we loop through the cells of our UI grid control and then place each value onto the observer using OnNext. When the process completes we then call the OnCompleted method of the observer to tell any subscribers that the process is finished.

Let’s look at the subscriber code

var tsk = new TaskCompletionSource<object>();

rowObservable.Buffer(20).
   SubscribeOn(SynchronizationContext.Current).
   ObserveOn(Scheduler.Default).
   Subscribe(r =>
   {
      foreach (var item in r)
      {
          // write each value to Excel (in this example)
      }
   }, () =>
   {
      tsk.SetResult(null);
   });

return tsk.Task;

In the above code we buffer pushed items from rowObserver so we only process every 20 items. We ObserveOn the default scheduler, so this will be a background thread (i.e. threadpool) but we SubscribeOn the current synchronization context – remember I mentioned this method should be called from the UI thread and hence the SubscribeOn (not ObserveOn) relates to the code we’re observing and this is on the UI thread. When the rowObserver completes we’ll still write out the last few items (if they’re less than the buffer size).

Note: It’s important to remember that SubscribeOn schedules the observable (in this case the UI code) not the code within the Subscribe method. This is scheduled using the ObserveOn method.

You’ll notice we use a puppet task, controlling a TaskCompletionSource and on completion of the rowObserver we set the result on the puppet task, thus allowing our method to be used in async/await scenarios.

Like I mentioned, this actually might not be the best solution to the original problem, but it was interesting getting it up and running.

I was actually looking through my past blog posts and noticed I’ve never published anything on Reactive Extensions (Rx). I actually have a draft “beginners” guide which I started a couple of years back, but never completed, so I may end up publishing that at a later date.

Note: The best resource for Rx information that I’ve found is Introduction to Rx, I’d definitely recommend you visit that site for complete and excellent documentation on Rx.

Okay, for this post I’m just going to focus on the FromEventPattern. Even though the syntax is pretty simple I have a habit of forgetting it everytime I come to use it, so as this blog is about refreshing my memory, I thought I’d write a blog post dedicated to this method.

Why use Rx instead of handling events in the conventional way?

One of the most obvious reasons to use Rx to handle events is that it can help stop those pesky memory leaks when we forget to unsubscribe to an event. Another reason is the IObservable returned from FromEventPattern is composable and we can filter events that we’re not interested in and/or marshal events from one thread onto another – for example when events are coming from a worker thread we might marshal them onto the UI thread.

Take a look at Observable.FromEventPattern Method to see the current syntax/overloads for this method.

I will not be covering all the overloads here but will instead look at the three I’ve used the most. So let’s get started…

FromEventPattern with an EventHandler which takes no EventArgs

In some cases you might have an event handler which doesn’t use EventArgs, for example

public event EventHandler Reverted;

in such a case we use the overload

public static IObservable<EventPattern<EventArgs>> FromEventPattern(
	Action<EventHandler> addHandler,
	Action<EventHandler> removeHandler)

So our code might look like this

var o = Observable.FromEventPattern(
   x => domainObject.Changed += x,
   x => domainObject.Changed -= x).
   Subscribe(_ => DoSomething());

FromEventPattern with standard event patterns (using the EventHandler class)

Building on the previous syntax…

In many cases, where the events that we want to subscribe to, use the “standard .NET event pattern”, (for example where the handler takes an EventArgs derived class) and were declared using the syntax

public event EventHandler<CellsInViewChangedEventArgs> CellsInViewChanged;

we will use this overload. Note: This version expects an EventHandler<> object.

public static IObservable<EventPattern<TEventArgs>> FromEventPattern<TEventArgs>(
	Action<EventHandler<TEventArgs>> addHandler,
	Action<EventHandler<TEventArgs>> removeHandler
)
where TEventArgs : EventArgs

To use this we simple supply the EventArgs derived type, for example

var o = Observable.FromEventPattern<CellsInViewChangedEventArgs>(
   x => grid.CellsInViewChanged += x,
   x => grid.CellsInViewChanged -= x).
   Subscribe(_ => DoSomething());

FromEvent pattern with standard event patterns (not using the Event Handler class)

Finally, for this post, in some cases we have events which do not use the EventHandler<> class, for example the WPF RoutedEvents use

public delegate void RoutedEventHandler(
   object sender, 
   System.Windows.RoutedEventArgs e)

In such cases we need to supply both the Delegate type and EventArgs type to the FromEventPattern, hence we use the syntax

public static IObservable<EventPattern<TEventArgs>> FromEventPattern<TDelegate, TEventArgs>(
	Action<TDelegate> addHandler,
	Action<TDelegate> removeHandler
)
where TEventArgs : EventArgs

our code will now look like this

var o = Observable.FromEventPattern<RoutedEventHandler, RoutedEventArgs>(
   x => button.Click += x,
   x => button.Click -= x).
   Subscribe(_ => DoSomething());

More…

I don’t intend to cover all the possible uses of the IObservable which is returned by the FromEventPattern, but suffice to say (as mentioned earlier). We can ensure events are marshaled onto the thread which created the observable, so for example if we create this from the UI thread we can write

var o = Observable.FromEventPattern<ValuesChangedEventArgs>(
   x => server.ValuesChanged += x,
   x => server.ValuesChanged -= x).
   ObserveOn(SynchronizationContext.Current).
   Subscribe(_ => DoSomething());

here, the ObserveOn will ensure that the events that might be coming from a worker thread are marshaled onto the current thread. Obviously if this was created on the UI thread this will marshal back onto that thread.

I often use the RX Throttle method, which is brilliant for scenario’s such as allowing a user to type text into a textbox and then only calling my code when the user pauses typing for an assigned amount of time. Or when scrolling up and down lists and when the user stops selecting items getting more data for the selected item from a webservice or the likes.

But what do we do about testing such interactions ?

Sidetrack

Instead of looking at the Throttle method, I needed to create a Throttle that actually buffers the data sent to it and when the timeout occurs gives the subscriber all the data not just the last piece of data (as Throttle would). So here’s some code I wrote (based heavily upon a StackOverflow post)

This post is meant to be about how we test such code, so let’s look at one of the unit test I have for this

[Fact]
public void ThrottleWithBuffer_EnsureValuesCorrectWhenOnNext()
{
   var scheduler = new TestScheduler();
   var expected = new[] { 'a', 'b', 'c' };
   var actual = new List<List<char>>();

   var subject = new Subject<char>();

   subject.AsObservable().
      ThrottleWithBuffer(TimeSpan.FromMilliseconds(500), scheduler).
      Subscribe(i => actual.Add(new List<char>(i)));

   scheduler.Schedule(TimeSpan.FromMilliseconds(100), () => subject.OnNext('a'));
   scheduler.Schedule(TimeSpan.FromMilliseconds(200), () => subject.OnNext('b'));
   scheduler.Schedule(TimeSpan.FromMilliseconds(300), () => subject.OnNext('c'));
   scheduler.Start();

   Assert.Equal(expected, actual[0]);
}

Now the key things here are the use of the TestScheduler which allows us the bend time (well not really, but it allows us to simulate the passing of time).

The TestScheduler is available in the Microsoft.Reactive.Testing.dll or from NuGet using Install-Package Rx-Testing

As you can see from the test, we now create the subscription to the observable and following that the scheduler is passed the “simulated” times. The first argument tells it the time an item is scheduled for, i.e. the first item is scheduled at 100ms, the next 200ms and finally 300ms. We then call Start on the scheduler to begin the time simulation.

Just think, if your code relied on something happening after a pause of a second then each test would have to wait one second before it could verify data, making a large test suit a lot slower to run. Using the TestScheduler we simply simulate a second passing.