Rx Reactive extensions: Unit testing with FromAsyncPattern

Question

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

var func = Observable.FromAsyncPattern<InType, OutType>(
  myWcfService.BeginDoStuff,
  myWcfService.EndDoStuff);

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):

[TestMethod()]
public void SomeTest()
{
   // arrange
   var container = GetAutoMockingContainer();

   container.Resolve<IMyWcfServiceClient>()
      .Expect(x => x.BeginDoStuff(null, null, null))
      .IgnoreArguments()
      .Do(
         new Func<Specification, AsyncCallback, object, IAsyncResult>((inData, asyncCallback, state) =>
            {
               return new CompletedAsyncResult(asyncCallback, state);
             }));

   container.Resolve<IRepositoryServiceClient>()
      .Expect(x => x.EndDoStuff(null))
      .IgnoreArguments()
      .Do(
         new Func<IAsyncResult, OutData>((ar) =>
         {
            return someMockData;
         }));

   // act
   var target = CreateTestSubject(container);

   target.DoMethodThatInvokesService();

   // 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:

• I have attempted to modify the second last line according to the suggestions here: Reactive Extensions Rx - unit testing something with ObserveOnDispatcher No love.

• I have added .Take(1) to the Rx code as follows:

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

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.

Update:

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.

Answer 1

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.

Answer 2

The problem is that MSTest.exe runs a Dispatcher (i.e. Dispatcher.Current != null), so ObserveOnDispatcher works. However, this Dispatcher does nothing! (i.e. queued dispatcher items will be ignored) Any code that you write that explicitly uses Schedule.Dispatcher is untestable

I solved this by brute-force in ReactiveXaml - here's the important bits:

http://github.com/xpaulbettsx/ReactiveXaml/blob/master/ReactiveXaml/Interfaces.cs#L44

We basically set up a global variable which defines the default scheduler, then we attempt to detect when we're in test runner.

http://github.com/xpaulbettsx/ReactiveXaml/blob/master/ReactiveXaml/ReactiveCommand.cs#L30

Then, in our classes that implement IObservable all take an IScheduler parameter, whose default value will end up being the global default scheduler. I could've probably done better, but this works for me and makes ViewModel code testable again.