Testing Concurrency in RxJava
Concurrency is hard enough — Testing it shouldn’t be.
While going about our day-to-day with RxJava, we’ll usually want to run a concurrent stream every now and then. But as with every intended action, we should test it.
Let’s dive into an example and test whether or not it’s actually running concurrently. Here’s a basic example in RxJava 2:
We have a list of 5 image IDs. We need to download their respective images from our ImageRepository dependency. In order to run this concurrently, we’ll flatMap the downloads together and subscribeOn a background thread for each download.
In order to simulate a delay, we will inject a DelayedRepository for our tests. We definitely don’t want to make actual network calls in our unit tests, so this is our “mock”:
Sequentially, if each download takes 5 seconds, loadImages() would take 25 seconds total. Downloading the images concurrently, however, should only take 5 seconds total.
This is how we will verify that our stream is running concurrency — with time.
Asserting Actual Time
If we block the downloading observable with .blockingGet(), we can simply assert that it takes less than 6 seconds (to account for JUnit boot-up time). This is a fairly straightforward approach:
Looks good! This will work, but a 5 second test is not a good unit test. Unit tests should run extremely quickly and this stream will produce a lot of extremely slow tests.
We could easily modify our DelayedRepository delay to 5 milliseconds instead of 5 seconds, but this would lessen the strength of our assertion and our impact.
We do care that our stream takes 5 seconds instead of 25 due to concurrency— therefore we test it.
We don’t care that our stream takes 25 milliseconds instead of 5 milliseconds if the stream actually runs sequentially. That’s not a big difference in time, so we shouldn’t waste resources testing that.
Instead of modifying our delay, we can fast-forward time so our tests run instantaneously.
Manipulating Virtual Time
In order to manipulate time, RxJava gives us the TestScheduler —a powerful tool that lets us simulate virtual time within our stream. Instead of using Schedulers.io() or Schedulers.computation() (the .delay() operator’s default scheduler), we will inject our stream with our own TestScheduler.
Let’s modify our ImageDownloader and DelayedRepository to prepare it for scheduler injection:
The only changes are really adding the ioScheduler and delayScheduler parameters to their respective classes. We use default parameters to optionally inject (in our new test) or not inject (in our previous test) our own schedulers.
Our new test looks like this:
Much better. Our new test now runs near-instantaneously, albeit having some start-up time for JUnit. Note that this is using the same ImageDownloader and DelayedRepository as the first test, the only difference being an injected TestScheduler.
Bonus: Sequential Streams
TestScheduler works well with testing delays in general. If you want to simulate a really long delay, feel free to do so — advance by the time you’re testing and assert the result you’re expecting. Here’s an updated ImageDownloader that loads using concatMap instead of flatMap and the same test structure to simulate 25 seconds instead of 5.
That’s pretty fast.
Github Repository
If you want to look at the code in its entirety, feel free to check out the repo below:
