Testing & error handling with Spring + Reactor

Introduction

As we saw in the previous post, the whole idea of reactive programming floats around two concepts: Data comes in asynchronous streams and those streams are immutable, so any modification results in new streams being created.
These facts makes the code not very intuitive and it is even worse when it comes to testing (even more if we are interacting with external resources, like a Web Service).
To help in the task, Spring and Reactor come with some handy tools that can be applied to simplify the development and testing.

First, let´s take a look at the scenario being tested:

Sequence diagram of the "production" code

Basically, we are requesting the StockPublisher to get some quotes and, in this case, they are retrieved from a Web Service hosted by Yahoo. The component tasked with getting the quotes uses a RestTemplate that finally makes the REST GET call.

REST testing with Spring´s MockRestServiceServer

As you can see in the diagram above, we are invoking a web service that might return different results (or might even not be available) during the testing time. In order to eliminate such variability, we are going to use this handy class provided by the Spring Framework: MockRestServiceServer.

Sequence diagram of the "test" code

The idea here is to have constant results as well as to introduce several test cases with errors, missing data, etc.
This is the test code:


I don´t want to go too deep on the MockRestServiceServer, but what we do with it is:

• Specify which invocations are we going to perform (HTTP method, URL, etc.)
• What we should expect in return (headers, HTTP statuses, etc.)
• We can even provide the response, so no invocation to the target URL is ever done!

The target here is to return controlled test data over which we could perform test assertions afterwards.

Reactive Junits with TestSubscriber

Now is time to talk about how to perform tests on our Flux/Mono streams using the TestSubscriber API.
As you can in the code above, we can subscribe to the Flux being tested and perform a large number of test assertions over it:

• Whether it has finished with an error or complete signal.
• How many values has the stream received? Which ones exactly?
• Also the TestSubscriber is a stream itself, so you can use it to launch new asynchronous events with the common onNext(..), onError(..), etc. methods.

Lets see an example of a test with three events correctly processed and an error raised while processing the fourth element:

Note: To ease up debugging and testing, the Reactor API comes with a log() method that creates a new Flux (remember, they are immutable) with the particular behavior that logs all the events received. This is the log trace:


Now, let´s give a fallback value of 0, if an error is thrown when parsing the String:



Why is the test still failing? Why is it yielding 3 instead of 12? Let´s see why.

Error handling

The API policy on errors is that they are terminal events. That is, a stream is terminated when a "Complete" or an "Error" signal is received.
Therefore, the only methods I have seen so far to handle errors are:

Credits to the Reactor Flux API docs.

Credits to the Reactor Flux API docs.

However, I am missing a method that allows handling discrete errors within the Flux without terminating it. Let´s take a look to the method being tested in the beginning of the post (and shown in the initial diagram):

As you can see we need a rather ugly piece that:

• Catches the exception
• Returns an empty Flux that will be merged with the rest of mapping results.

Summary

I´m still exploring the API to see if there is a better way to handle the errors (I even opened a ticket in the Reactor project issue tracker), and I have questions for you:

• Do you have any feedback using Reactor? Would you accomplish this in other way?
• Do you feel that is mature enough? 
• Do you use any of the other modules (I/O, the EventBus, etc.) 

Feel free to share your thoughts!