RxJs Testing in Real World Applications

ReactiveX (and it's JavaScript implementation RxJS) is a fantastic technology. We use it heavily to handle the data streams between view models, our REST API and the local IndexedDb data storage in the data model of our web client. But also for the more complex UI events in our AngularJS web app, such as the user search. Especially in our user search the debounce() operator helps us to keep the number of search requests made to the server low and still have a real time search feel. But RxJS has its flaws too, especially when testing existing code.

The problem with RxJS and ReactiveX in general is its overly complex written documentation. The entry threshold therefore is very high. But when you get to the point where you understand the concept and the different operators a little better, these problems minimize. And then the next problem of the documentation arrises: unit testing.

Testing code all inside the test specs?

The dilemma is the way in which the RxJS documentation describes testing your RxJS application. All the examples have the testable code inside the tests itself. This way you can create streams using a VirtualScheduler like the TestScheduler or using the test/mock implementations of subscribe(), onNext() and onComplete(), everything fine.

But you usually don't want to rebuild your logic inside your tests so you can maintain two versions of your code with a lot of seesaw changes.
Besides all the work that comes with this, the approach also is very error-prone! Copy the original code wrong oder forget to transfer changes to the test and your test tells you, everything is ok when it it's not.
Unfortunately there is little information about testing existing code when searching the web.

When I was building our web client's data model (that was the first time I used RxJS in a bigger scope) I stumbled on that problem: How can I use the TestScheduler with my existing code? There doesn't seem to be a possibility to change the default scheduler to the test scheduler other than overwriting the Rx.Scheduler.default value. Even then there are other schedulers used as default for different operators.

Using the Scheduler

When testing your asynchonous RxJS code, the method you need the most is the scheduleAbsolute() method to test your code at different moments:

it('should test your code', function(done) {  
  var scheduler = new TestScheduler();

  // Test code at 250ms after observable creation
  scheduler.scheduleAbsolute(null, 250, function() {

    // you can also put in some test data 
    // in the middle of runtime to test your code

  // after 900ms
  scheduler.scheduleAbsolute(null, 900, function() {

    // stop the scheduler and call done() if are finished testing at that point

  // start the scheduler, otherwise no async code is executed

Always think to stop the scheduler. Especially when you want to test just a certain part of your code. Otherwise you could get different return values than you expect.

Injecting the TestScheduler to operators

After some search I stumbled on a post from the blog of Emmanuel Kong. You can tell time related operations which scheduler they should use by passing the scheduler as an argument. The problem with that approach is that this also has to be done inside the test itself or you have to modify your code to pass a scheduler and allow to change it from outside:

function Class(some, args, scheduler) {  
  var _scheduler = scheduler || Rx.Scheduler.default;
  this.stream = new Rx.Subject();
  this.stream.debounce(250, _scheduler).subscribe(function(item) {
    // do something

This way you can inject the scheduler into your code, but this also means that you have to do this for every class, every file, every snippet. Some operators also use other schedulers than the default one and most of the time you don't really want to mess with the schedulers. So there must be another way to inject your scheduler.

Using spies to inject your Scheduler

When it comes to testing JavaScript code (for the frontend) I trust in Jasmine. Jasmine already has build in spies and so I don't have to use Sinon (which also is a great library, we use it for our backend tests together with Mocha).

Spies are perfect for injecting the TestScheduler into existing code. On top of that you get the possibility to check if your operators are called with the right arguments and in the correct quantity. The only problem is, that you only can spy on methods of existing objects, and thanks to chaining your filters, mapping functions and other operators it isn't possible to access the right object from outside to spy on its operator(s).

Because of that you have to spy on the operators in the prototype object of the Rx.Observer class. This way you can spy on every execution of the operator you want to spy on:

var scheduler = new Rx.TestScheduler();  
var originalThrottle = Rx.Observable.prototype.throttle;  
spyOn(Rx.Observable.prototype, 'throttle').and.callFake(function(dueTime) {  
  return originalThrottle.call(this, dueTime, scheduler);

Here's how it works: First you have to backup your original function in a variable (to call it later). After that you can spy on the operator you want to inject your scheduler into (in the case above the throttle() operator, because it has static arguments – more on that later).

You have to chain the spy with .and.callFake() to replace the original behavior. Inside this you have to call the original throttle() and return it's values. Be sure to call the original operator with the .call() method because all the operators of Rx.Observable are instance methods which are called with the this context of the Rx.Obserable instance.

Beware of exceptions

The throttle() operator is fairly easy to spy on because it only has one argument besides the scheduler to use and both arguments always have the same semantics. Some operators have arguments which meaning depend on the type of the arguments (like timer(), delay(), timeout() and debounce()).

For example: When spying on the debounce() operator, you have to look for the first argument. It can be either be a function or a number. When it is a function it only uses the first argument and does not use the scheduler, when it's a number the debounce() operator uses the scheduler passed in the second argument. This exceptions need special treatment:

var scheduler = new Rx.TestScheduler();  
var originalDebounce = Rx.Observable.prototype.debounce;  
spyOn(Rx.Observable.prototype, 'debounce').and.callFake(function(arg1) {  
  if (typeof arg1 === 'number') {
    return originalDebounce.call(this, arg1, schedulerInstance);
  return originalDebounce.call(this, arg1);

Other exceptions are the timer() and interval() operators, which are not instance methods and therefore are contained in the Rx.Observable object instead of the Rx.Observable.prototype.

There's an utility for that

Because we have to inject the TestScheduler into most of our test code, I have decided to write a little utility that does that using the spies of the Jasmine framework. It is called RxJS-TestScheduler-Injector and is available through GitHub.

You just have to install it through Bower using bower install rxjs-testscheduler-injector --save-dev (or just by downloading it from the GitHub and handle it manually).

Using RxJS-TestScheduler-Injector

All you have to do now is to use following method:

var scheduler = new Rx.TestScheduler();  

The utility injects the scheduler into most of the operators that accept schedulers passed through an argument (if I have forgotten one, please let me know (or fork the project and add it yourself) - I am glad for every feedback or pull request on this.

If you just want to inject the scheduler into one specific operator, you can do it with the injectInto() method:

RxJsTestSchedulerInjector.injectInto('debounce', scheduler);  

You can learn more about the usage of the utility in the readme on GitHub.

Thats it! Do you have another method for working with the TestScheduler or testing RxJS apps in general? Let us know in the comments!

comments powered by Disqus