How to test RxJava code

Assuming you've read about RxJava and have experienced examples like those in "RxJava Getting Started: Example Analysis", you're now ready to explore responsive programming in your own code. However, you're still wondering how to test the new features you might find in your codebase. Let's explore how to test RxJava code.

[[178035]]

Key points of this article:

• RxJava has built-in, test-friendly solutions.
• Use TestSubscriber to verify Observable.
• Use TestScheduler to achieve strict control of time.
• The Awaitility library provides further control over the testing environment.

With responsive programming, we have to change the way we reason about a given problem, because we have to focus on the flow of data as a stream of events, rather than individual data items. Events are usually produced and consumed by different threads, so we must have a clear understanding of concurrency issues when writing tests. Fortunately, RxJava provides built-in support for testing Observable and Subscription, and it is built directly into the core dependencies of RxJava.

***step

Let's review the vocabulary example given in the article "RxJava Introduction: Example Analysis" and see how to test this example. Let's start with the setting of the basic test tool. In our test architecture, JUnit is used as the test tool.

In fact, if no scheduler is given, Subscription will run on the calling thread by default. Therefore, we will use the native method in the *** test. This means that we can implement an object of the Subscription interface and assert its status immediately after the Subscription occurs.

Note the use of an explicit List<String> container here, along with the actual subscribers to accumulate the results. Given the simplicity of the given test, you might think that this explicit accumulator approach is good enough. But remember that production-grade Observables may encapsulate errors or may produce unexpected events. The simple combination of Subscriber and accumulator in the example is not enough to cover this situation. But don't worry, RxJava provides the TestSubscriber type to handle this situation. Let's refactor the above test using the TestSubscriber type.

TestSubscriber not only replaces the user accumulator, but also provides some additional behaviors. For example, it can give the size of the received messages and the data associated with each event. It can also assert that the Subscription is completed and no errors occur during the consumption of the Observable. Although the Observable in the current test does not generate any errors, going back to the article "RxJava Introduction: Example Analysis", we learned that Observable treats exceptions and data events the same. We can simulate errors by connecting exception events in the following way:

All of this works well in the limited use cases we've given. But actual production code may look completely different from the examples. So in the following, we'll consider some more complex production examples.

Custom Scheduler

In production code, many use cases of Observable are executed on a specific thread, which is called a "Scheduler" in the context of responsive programming. Many Observable operations use an optional scheduler parameter as an additional parameter. RxJava defines a series of named schedulers that are always available, including the IO scheduler (io), the computation scheduler (computation, which is a shared thread), and the new thread scheduler (newThread). Developers can also implement their own custom schedulers. Let's modify the Observable code by specifying the computation scheduler.

When you run this code, you will immediately find that it is problematic. The Subscriber executes its assertions on the test thread, but the Observable generates values ​​on a background thread (the computation thread). This means that the Subscriber assertions may be executed before the Observable generates all relevant events, thus causing the test to fail.

To ensure the smooth execution of the test, there are some strategies to choose from:

• Converts an Observable to a blocking one.
• Forces the test to wait until a given condition is met.
• Converts a compute scheduler to an immediate (Schedulers.immediate()) scheduler.

We will go into more detail about each strategy, but we will start with "Converting Observables to Blocking" because it requires the least technical work to implement, regardless of the scheduler used. We will assume that the data is generated in a background thread, which will cause the Subscriber to be notified from the same background thread.

What we want to do is force all events to be generated and the Observable to complete in the test before the next statement is executed. This is done by calling the toBlocking() method on the Observable itself.

Although this method is suitable for the simple code we have given, it may not be suitable for actual production code. What will happen if it takes a long time for the producer to generate all the data? This will make the test very slow, increase the compilation time, and there may be other performance issues. Here I recommend a convenient library, Awaitility (https://github.com/awaitility/awaitility). Simply put, Awaitility is a DSL that expresses expectations related to asynchronous systems in a precise, simple and easy-to-read way. You can use Maven to add Awaitility dependencies in your project.

 <dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
 <version>2.0.0</version>
 <scope>test</scope>
 </dependency>

Or using Gradle:

 testCompile 'org.awaitility:awaitility:2.0.0'  

The entry point of Awaitility DSL is org.awaitility.Awaitility.await() method (see lines 13 and 14 in the example below). You can use Awaitility to define conditions that must be met for the test to continue. You can also add timeouts or other timing constraints to the conditions, such as minimum, maximum or duration ranges. For the above example, the following code shows how to use Awaitility in the result:

This version of the test does not change the nature of the Observable in any way, which allows you to test it without making any changes to your production code. This version of the test uses a maximum of 2 seconds of waiting time to let the Observable do its job by checking the Subscriber state. If all goes well, the Subscriber state is released to all 9 events within 2 seconds.

Awaitility has good collaboration with Hamcrest matchers, Java 8 lambda expressions and method references, giving precise and readable test conditions. Awaitility also provides pre-made extensions for widely used JVM languages, including Groovy and Scala.

The last strategy we are going to present uses the RxJava extension mechanism, which is released as part of the RxJava API. RxJava defines a series of extension points that allow fine-tuning of almost any default RxJava behavior. This extension mechanism allows us to provide modified values ​​for specific RxJava features. Using this mechanism, we can inject the selected scheduler in the test without having to care about the scheduler specified in the generated code. This is exactly the method we are looking for, which is encapsulated in the RxJavaHooks class. Assuming that the product code depends on the calculation scheduler, we will override its default value and return a scheduler that acts as the called code to make the event processing happen. This is the immediate scheduler (Schedulers.immediate()). Here is the code for the test:

In the test, the production code is unaware that the calculation scheduler is immediate. Note that the hook function must be reset, otherwise the immediate scheduler settings may leak, causing tests everywhere to be broken. Using try/finally code blocks will obscure the purpose of the test to a certain extent, but fortunately we can use JUnit rules to refactor this behavior to make the test more concise and the result more readable. Here is a possible implementation code using the above rules:

In addition, we have rewritten the generation methods of the other two schedulers. This rule is more general for other test targets in the future. In the new test case class, the use of this rule is very direct. Just simply define a domain and annotate the new type as @Rule. The sample code is as follows:

Finally, we get the same behavior as in the previous test, but without the clutter. Let's take a moment to review what we have done so far:

• Subscribers will process data in the same thread as long as no special scheduler is used. This means that after a Subscriber subscribes to an Observable, we can make assertions on the Subscriber.
• TestSubscriber can accumulate events and make additional assertions about its own status.
• Any Observable can be converted to blocking, which allows us to synchronously wait for events to be generated regardless of the scheduler used by the Observable.
• RxJava provides an extension mechanism that allows developers to override its default methods and inject them into product code in an appropriate manner.
• Concurrent code can be tested using the Awaitility DSL.

Each of the above techniques works in different scenarios, but all are related by a common thread: the test code needs to wait for the Observable to complete before making assertions about the Subscriber state. Given that the behavior of the Observable generates data, is there a way to inspect that behavior? In other words, is it possible to do live debugging of the Observable programmatically? We will provide such a technique later in this article.

Manipulate time

So far we have been testing Observables and Subscriptions in a black-box way. Next we will consider another technique for manipulating time, which allows us to look under the hood and see the Subscriber state while the Observable is still active. In other words, we will use a white-box testing technique that uses RxJava's TestScheduler class, and this is where RxJava comes to the rescue. This special scheduler can precisely set the internal use of time, such as advancing the time by half a second, or making the time jump by 5 seconds. We will first show how to create an instance of this new scheduler, and then discuss testing the code.

The "production" code has changed slightly, as we now use a method that binds to the Scheduler's interval() to generate the count (line 6) rather than generating a range of counts. This has the side effect that the counts start at zero rather than 1. Once the Observable and test Scheduler are configured, we immediately make the assertion that the Subscriber has no value (line 15) and has not completed or generated any errors (line 16). This is a sanity test, as the Scheduler has not been moved at this point, and therefore no values ​​have been produced by the Observable or received by the Subscriber.

Next, we advance the time by one full second (line 19), which will cause the Observable to produce its first value, which is what the subsequent set of assertions checks (lines 22-24).

Next we advance the time from the current time to 9 seconds. Note that this means advancing the time to exactly 9 seconds after the scheduler was started (not advancing 1 second then 9 seconds, which would be 10 seconds after the scheduler check was started). In other words, the advanceTimeBy() method advances the scheduler's time relative to the current position, while advanceTimeTo() advances the time in an absolute manner. We then make the next round of assertions (lines 28-20) to ensure that all data is generated by the Observable and consumed by the Subscriber. Another thing to note is that when using the TestScheduler, the actual time is advanced immediately, which means that the test does not have to actually wait 9 seconds to complete.

As you can see, using this scheduler is very convenient, just provide it to the Observable you are testing. However, it does not work well with Observables that use a specific type of scheduler. But wait, previously we saw how to use RxJavaHooks to switch a scheduler without affecting production code, this time providing a TestScheduler that replaces the immediate scheduler (lines 13-15). We can even apply the same technique with custom JUnit rules, so that the previous code can be rewritten in a more reusable way. First, the new rule is:

Next comes the actual test code (in a new test case class) to use our test rules:

There you have it. Using the method of injecting the TestScheduler via RxJavaHooks, you can write test code without changing the original Observable composition, and it also gives you a way to change the time during the execution of the observable itself and make assertions at specific points. All these techniques given in this article should be enough for your chosen RxJava code to test.

future

RxJava is one of the most popular libraries that provide responsive programming capabilities for Java. The upcoming version 2.0 will be a redesign to make the RxJava API better conform to the Reactive Streams specification. The Reactive Streams specification targets Java and JavaScript runtimes and provides a standard for asynchronous stream processing using non-blocking back pressure. This means that there will be some API improvements in the next version of RxJava. For a detailed description of these improvements, see the RxJava wiki.

For testing, the core types (Observable, Maybe, and Single) now have convenient test() methods that create TestSubscriber instances on the spot. It is also possible to chain method calls on TestSubscriber, and there are some new assertion methods for this type of usage.

Original author: Andres Almiray