Andrew Anderson Andrew Anderson - 1 month ago 7x
C# Question

Unit testing with FromAsyncPattern

The Reactive Extensions have a sexy little hook to simplify calling async methods:

var func = Observable.FromAsyncPattern<InType, OutType>(

func(inData).ObserveOnDispatcher().Subscribe(x => Foo(x));

I am using this in an WPF project, and it works great at runtime.

Unfortunately, when trying to unit test methods that use this technique I am experiencing random failures. ~3 out of every five executions of a test that contain this code fails.

Here is a sample test (implemented using a Rhino/unity auto-mocking container):

public void SomeTest()
// arrange
var container = GetAutoMockingContainer();

.Expect(x => x.BeginDoStuff(null, null, null))
new Func<Specification, AsyncCallback, object, IAsyncResult>((inData, asyncCallback, state) =>
return new CompletedAsyncResult(asyncCallback, state);

.Expect(x => x.EndDoStuff(null))
new Func<IAsyncResult, OutData>((ar) =>
return someMockData;

// act
var target = CreateTestSubject(container);


// Run the dispatcher for everything over background priority
Dispatcher.CurrentDispatcher.Invoke(DispatcherPriority.Background, new Action(() => { }));

// assert
Assert.IsTrue(my operation ran as expected);

The problem that I see is that the code that I specified to run when the async action completed (in this case, Foo(x)), is never called. I can verify this by setting breakpoints in Foo and observing that they are never reached. Further, I can force a long delay after calling DoMethodThatInvokesService (which kicks off the async call), and the code is still never run. I do know that the lines of code invoking the Rx framework were called.

Other things I've tried:

This improved my failure rate to something like 1 in 5, but they still occurred.

  • I have rewritten the Rx code to use the plain jane Async pattern. This works, however my developer ego really would love to use Rx instead of boring old begin/end.

In the end I do have a work around in hand (i.e. don't use Rx), however I feel that it is not ideal. If anyone has ran into this problem in the past and found a solution, I'd dearly love to hear it.


I also posted on the Rx forums, and they will be including a test scheduler with a coming release. That will probably be the ultimate solution once it is available.


The problem is caused by the asynchronous nature of the calls scheduled by ObserveOnDispatcher. You can't guarantee that they are all completed by the time your test finishes. So you need to bring the scheduling under your control.

How about injecting the scheduler into your class?

Then, rather than calling ObserveOnDispatcher, you call ObserveOn, passing in the the IScheduler implementation that was injected.

At run-time you would inject the DispatcherScheduler, but in your tests you would inject a fake scheduler that queues up all the actions that it is given and runs them at a time controlled by your tests.

If you don't like the idea of having to inject a scheduler everywhere that you're using Rx, how about creating your own extension method, something like this (untested code ahead):

public static MyObservableExtensions
   public static IScheduler UISafeScheduler {get;set;}

   public static IObservable<TSource> ObserveOnUISafeScheduler(this IObservable<TSource> source)
       if (UISafeScheduler == null) 
          throw new InvalidOperation("UISafeScheduler has not been initialised");

       return source.ObserveOn(UISafeScheduler);

Then at run-time, initialise UISafeScheduler with DispatcherScheduler, and in your tests, intialise it with your fake scheduler.