Let me boil down your question to its essentials:
Is there a way, using the TestScheduler
, to execute a reactive pipeline and wait for its completion even when it contains asynchronous calls?
I should warn you up front, there is no quick and easy answer here, no convenient "trick" that can be deployed.
Asynchronous Calls and Schedulers
To answer this question I think we need to clarify some points. The term "asynchronous call" in the question above seems to be used specifically to refer to methods with a Task
or Task<T>
signature - i.e. methods that use the Task Parallel Library (TPL) to run asynchronously.
This is important to note because Reactive Extensions (Rx) takes a different approach to handling asynchronous operations.
In Rx the introduction of concurrency is managed via a scheduler, a type implementing the IScheduler
interface. Any operation that introduces concurrency should make a available a scheduler parameter so that the caller can decide an appropriate scheduler. The core library slavishly adheres to this principle. So, for example, Delay
allows specification of a scheduler but Where
does not.
As you can see from the source, IScheduler
provides a number of Schedule
overloads. Operations requiring concurrency use these to schedule execution of work. Exactly how that work is executed is deferred completely to the scheduler. This is the power of the scheduler abstraction.
Rx operations introducing concurrency generally provide overloads that allow the scheduler to be omitted, and in that case select a sensible default. This is important to note, because if you want your code to be testable via the use of TestScheduler
you must use a TestScheduler
for all operations that introduce concurrency. A rogue method that doesn't allow this, could scupper your testing efforts.
TPL Scheduling Abstraction
The TPL has it's own abstraction to handle concurrency: The TaskScheduler
. The idea is very similar. You can read about it here..
There are two very important differences between the two abstractions:
- Rx schedulers have a first class representation of their own notion of time - the
Now
property. TPL schedulers do not.
- The use of custom schedulers in the TPL is much less prevalent, and there is no equivalent best practice of providing overloads for providing specific
TaskSchedulers
to a method introducing concurrency (returning a Task
or Task<T>
). The vast majority of Task
returning methods assume use of the default TaskScheduler
and give you no choice about where work is run.
Motivation for TestScheduler
The motivation to use a TestScheduler
is generally two-fold:
- To remove the need to "wait" for operations by speeding up time.
- To check that events occurred at expected points in time.
The way this works depends entirely on the fact that schedulers have their own notion of time. Every time an operation is scheduled via an IScheduler
, we specify when it must execute - either as soon as possible, or at a specific time in the future. The scheduler then queues work for execution and will execute it when the specified time (according to the scheduler itself) is reached.
When you call Start
on the TestScheduler
, it works by emptying the queue of all operations with execution times at or before its current notion of Now
- and then advancing its clock to the next scheduled work time and repeating until its queue is empty.
This allows neat tricks like being able to test that an operation will never result in an event! If using real time this would be a challenging task, but with virtual time it's easy - once the scheduler queue is completely empty, then the TestScheduler
concludes that no further events will ever happen - since if nothing is left on its queue, there is nothing there to schedule further tasks. In fact Start
returns at this precisely this point. For this to work, clearly all concurrent operations to be measured must be scheduled on the TestScheduler
.
A custom operator that carelessly makes its own choice of scheduler without allowing that choice to be overriden, or an operation that uses its own form of concurrency without a notion of time (such as TPL based calls) will make it difficult, if not impossible, to control execution via a TestScheduler
.
If you have an asynchronous operation run by other means, judicious use of the AdvanceTo
and AdvanceBy
methods of the TestScheduler
can allow you to coordinate with that foreign source of concurrency - but the extent to which this is achievable depends on the control afforded by that foreign source.
In the case of the TPL, you do know when a task is done - which does allow the use of waits and timeouts in tests, as ugly as these can be. Through the use of TaskCompleteSources
(TCS) you can mock tasks and use AdvanceTo
to hit specific points and complete TCSs, but there is no one simple approach here. Often you just have to resort to ugly waits and timeouts because you don't have sufficient control over foreign concurrency.
Rx is generally free-threaded and tries to avoid introducing concurrency wherever possible. Conversely, it's quite possible that different operations within an Rx call chain will need different types of scheduler abstraction. It's not always possible to simulate a call chain with a single test scheduler. Certainly, I have had cause to use multiple TestSchedulers
to simulate some complex scenarios - e.g. chains that use the DispatcherScheduler
and TaskScheduler
sometimes need complex coordination that means you can't simply serialize their operations on to one TestScheduler
.
Some projects I have worked on have mandated the use of Rx for all concurrency specifically to avoid these problems. That is not always feasible, and even in these cases, some use of TPL is generally inevitable.
One particular pain point
One particular pain point of Rx that leaves many testers scratching their heads, is the fact that the TPL -> Rx family of conversions introduce concurrency. e.g. ToObservable
, SelectMany
's overload accepting Task<T>
etc. don't provide overloads with a scheduler and insidiously force you off the TestScheduler
thread, even if mocking with TCS. For all the pain this causes in testing alone, I consider this a bug. You can read all about this here - dig through and you'll find Dave Sexton's proposed fix, which provides an overload for specifying a scheduler, and is under consideration for inclusion. You may want to look into that pull request.
A Potential Workaround
If you can edit your code to use it, the following helper method might be of use. It converts a task to an observable that will run on the TestScheduler and complete at the correct virtual time.
It schedules work on the TestScheduler that is responsible for collecting the task result - at the virtual time we state the task should complete. The work itself blocks until the task result is available - allowing the TPL task to run for however long it takes, or until a real amount of specified time has passed in which case a TimeoutException
is thrown.
The effect of blocking the work means that the TestScheduler
won't advance its virtual time past the expected virtual completion time of the task until the task has actually completed. This way, the rest of the Rx chain can run in full-speed virtual time and we only wait on the TPL task, pausing the rest of the chain at the task completion virtual time whilst this happens.
Crucially, other concurrent Rx operations scheduled to run in between the start virtual time of the Task based operation and the stated end virtual time of the Task are not blocked and their virtual completion time will be unaffected.
So set duration
to the length of virtual time you want the task to appear to have taken. The result will then be collected at whatever the virtual time is when the task is started, plus the duration specified.
Set timeout
to the actual time you will allow the task to take. If it takes longer, a timeout exception is thrown:
public static IObservable<T> ToTestScheduledObseravble<T>(
this Task<T> task,
TestScheduler scheduler,
TimeSpan duration,
TimeSpan? timeout = null)
{
timeout = timeout ?? TimeSpan.FromSeconds(100);
var subject = Subject.Synchronize(new AsyncSubject<T>(), scheduler);
scheduler.Schedule<Task<T>>(task, duration,
(s, t) => {
if (!task.Wait(timeout.Value))
{
subject.OnError(
new TimeoutException(
"Task duration too long"));
}
else
{
switch (task.Status)
{
case TaskStatus.RanToCompletion:
subject.OnNext(task.Result);
subject.OnCompleted();
break;
case TaskStatus.Faulted:
subject.OnError(task.Exception.InnerException);
break;
case TaskStatus.Canceled:
subject.OnError(new TaskCanceledException(task));
break;
}
}
return Disposable.Empty;
});
return subject.AsObservable();
}
Usage in your code would be like this, and your assert will pass:
Observable
.Return(1)
.Select(i => Whatever().ToTestScheduledObseravble(
scheduler, TimeSpan.FromSeconds(1)))
.Concat()
.Subscribe(_ => Interlocked.Increment(ref count));
Conclusion
In summary, you haven't missed any convenient trick. You need to think about how Rx works, and how the TPL works and decide whether:
- You avoid mixing TPL and Rx
- You mock your interface between TPL and Rx (using TCS or similar), so you test each independently
- You live with ugly waits and timeouts and abandon the
TestScheduler
altogether
- You mix ugly waits and timeouts with
TestScheduler
to bring some modicum of control over your tests.