Common rxjs mistakes

This is a follow-up to my post about common rxjs patterns. In that post I summarised some of the common patterns that people look to express using observables. When looking at the questions people ask about rxjs, it’s clear that they’re not taking advantage of the full power of the library. In this post I’d like to highlight ways people can use rxjs to its fullest.

Subscriptions within subscriptions

One of the biggest selling points of reactive extensions is the ability to create quite complex observables, by composing smaller, well-defined operators together. Often what we’re trying to do has already been done 1000 times before, and so the operators come pre-packaged with the library. If you’re unsure of what to use, this decision tree can aid your thinking by steering you to a solution.

What I see people trying to do instead of composing these operators, is to simply subscribe to an observable A$, and subscribe to a second observable B$ within the callback to A$. Unfortunately this often neglects what we get for free when composing operators, namely the propagation of next, complete, error and unsubscription.

In example above, A$ produces an error. What would you expect to happen? I would probably expect B$ to be unsubscribed from, but this isn’t what happens. I’ve changed the example to match my expectation:

The observable really is a pipe, with next => next, error => error, complete => complete and unsubscribe <= unsubscribe.

In addition, these operators control the order in which observables are subscribed, and hence in which order values should be emitted. To give you a flavour, there’s switchMap, concatMap, mergeMap, exhaustMap, which each have different characteristics. You lose control of this when subscribing manually, so the solution is to build your pipe by composing operators, and then subscribe once at the end of your pipe, which is where you want to start producing results.

Using Subjects and poking values

How would you create a collection containing the numbers 1, 2 and 3, and then print the values of the collection?

This approach seems reasonable. Create an array, poke the values and then iterate over it. So how would you create an observable that emits the numbers 1, 2 and 3, and then print those values?

This is the solution that a lot of people newly exposed to reactive extensions would write. There are several issues with it:

1. You have to subscribe before calling next. If you’re late, you won’t be seeing those values again!
2. If there’s an error, how do you retry?
3. The subject, or a public method if the subject is private, is often exposed in order to poke values into it. It’s difficult to track what is providing input, and also at which time.
4. Try writing marble tests with subjects. Not only is it more difficult in general, but there may also be cases that are impossible to test.

Here’s an example using the reactive way:

The first example uses a standard observable-creation function provided by the library. Only if such functionality doesn’t already exist (one might have to integrate with a 3rd-party API with awkward methods), then try creating a new Observable. This is analogous to creating a Promise, where functions are provided to be invoked when the work has been completed and a value is ready to be emitted. The difference is that the observable uses lazy evaluation — the producer function will only be invoked on subscription. By this token it’s also possible to retry on error, namely because the producer function will once again be invoked.

Not writing unit tests

If you’ve followed the previous steps then you should be in a good position to write unit tests for your observables. The library comes with some utilities for writing declarative expectations of what your observables should produce: marble tests. The idea behind it is as follows:

Imagine if you were to test a collection of values. Using an assertion library, you would write something similar to: expect([1,2,3]).toEqual([1,2,3]). I think this is quite clear to read — at each index of the array under test you expect to see a particular value. The principle when testing observables is the same, except that the index is the time at which the value was emitted.

Here’s the sample for the operator race, which takes an arbitrary number of observables and then mirrors the one that’s fastest to emit an initial value. And here’s the tests I’ve come up with to show that it works:

By swapping out the input observables with those provided by the test scheduler, one can ensure that the test is run synchronously, deterministically and also that they’re unsubscribed.

Forgetting to signal completion when testing concatenated observables

Let’s say you want to concatenate two arrays: […arrayA, …arrayB]. It’s very straightforward. Concatenating two observables is also straightforward: concat(observableA, observableB). So why aren’t the values from observableB emitted in my unit test?

Let’s take a look at an example that fetches all existing items (doesn’t matter what they are) from the API, and then subscribes to a realtime stream of their updated values:

The test is mocking the API client to provide cold observables created via the test scheduler. In reality, the client would be some kind of http client, perhaps using the fetch API. In cases where there’s a request and corresponding response, it’s clear that the observable will produce a value and then complete, or produce an error if something went wrong.

It’s easy to forget that observables can stream values forever when one normally uses them in the same situations that a Promise could also be used for. The solution in this case is quite simple: signal completion of the observable by using the | character:

Using Promises and wondering why unit tests are failing

It’s important to remember that Promise callbacks are always executed asynchronously. This is in contrast to observables, where the scheduler decides how to execute callback functions and in marble tests they’re executed synchronously.

What effect does this have on our unit tests? It means that we would need to somehow await the result. Some people write marble tests where they pass in a resolved Promise and then wonder why the output result is empty. The reason is that even though it looks like the Promise is resolved, the callback will still be executed asynchronously. The items specified in the marble diagram “- — — — a — b” are emitted synchronously, and then the Promise callback will be executed. This completely destroys the ordering of the emissions.

How can we fix this? I suggest staying entirely within the world of observables. Create simple wrappers around APIs that return Promises and convert them to observables instead. In your tests you can easily mock these and have that sweet, sweet, synchronous behaviour that we all long for.