Reactor

Getting Started

If you have a Maven build, add the following to your pom file:

<groupId>io.projectreactor</groupId>

<artifactId>reactor-core</artifactId>

<version>3.1.9.RELEASE</version>

</dependency>

<groupId>io.projectreactor</groupId>

<artifactId>reactor-test</artifactId>

<version>3.1.9.RELEASE</version>

</dependency>

For Gradle builds, add the following to your Gradle build file’s dependencies:

compile 'io.projectreactor:reactor-core:3.1.9.RELEASE'

testCompile 'io.projectreactor:reactor-test:3.1.9.RELEASE'

Flux

Flux<T> is the main entry point for Reactor reactive streams and is similar to RxJava’s Observable. Mono<T> is like a Flux but for zero to one element. Both Mono and Flux implement org.reactivestreams.Publisher .

import reactor.core.publisher.Flux;

import reactor.core.publisher.Mono;

Much like in RxJava, Reactor uses Schedulers to decide on what thread to run.

For example, you might create a range like the following and publish on “Schedulers.parallel()” which provides a thread cache for executing in parallel:

Flux.range(1, 100)

.publishOn(Schedulers.parallel())

.subscribe(v -> System.out.println(v));

The preceding code would print out the numbers 1 through 100.

Handling errors in Reactor is also very similar to RxJava. The following methods may be used on a Flux or Mono (generic types omitted for brevity):

onErrorResume(Function) : Takes the exception and returns a different Publisher as a fallback or secondary stream.
onErrorMap(Function) : Takes the exception and allows you to modify it or return a completely new Exception if you prefer.
onErrorReturn(T) : Provides a default value to use when an error arises.
doOnError(Consumer<? super Throwable>) : Allows you to handle the error without effecting the underlying stream in any way.

Errors are always ending events for a Flux or Mono and should be handled by the Subscriber. However, many times, such as in the preceding example, an error is not possible and therefore does not need to be handled.

Mono

Mono is much like a Flux but for just one or zero elements. Think of it like a translation of Java 8’s Optional class into the Reactive Streams world. For example, the following would print out the value “hello”:

Mono.just("hello").subscribe(v -> System.out.println(v));

Mono is very similar to Flux except that it has methods like

justOrEmpty(T) : Takes a nullable value and converts into a Mono. If null, the result is the same as Mono.empty().
justOrEmpty(Optional) : Takes an Optional and converts into a Mono directly.

Unlike Java’s Optional, Mono can handle errors, among other things. For example, a method that returns Mono might do the following:

return Mono.error(new RuntimeException("your error"))

The corresponding code can handle errors from a Mono in the same way as with a Flux (using onErrorResume, onErrorMap, or onErrorReturn).

Creating a Flux or Mono

You can create a Flux from fixed data (cold) or programmatically from dynamic data (hot).

The following are some different ways to create a cold Flux:

Flux<String> flux1 = Flux.just("a", "b", "foobar"); //1

List<String> iterable = Arrays.asList("a", "b", "foobar");

Flux<String> flux2 = Flux.fromIterable(iterable); //2

Flux<Integer> numbers = Flux.range(1, 64); //3

1.
Create a Flux from a list of values.
2.
Create a Flux from an Iterable.
3.
Create a range from 1 to 64.
Here’s how to create a simple Mono:
Mono<String> noData = Mono.empty(); //1
Mono<String> data = Mono.just("foo"); //2
4.
Create an empty Mono.
5.
Create a Mono with one element.

You can programmatically create a hot or cold Flux using one of the generate, create, or push methods . If the data is of a continuous nature, such as user input, a WebSocket, or network packets, it would be considered hot.

The generate method (in one variety) takes a Supplier and a BiFunction. The function takes as parameters the current state and a SynchronousSink<T> which can be used to publish the next state of the stream. For example, the following uses an AtomicLong instance to increment the numbers 0 through 10 and supplies the square of each number:

Flux<Long> squares = Flux.generate(

AtomicLong::new, //1

(state, sink) -> {

long i = state.getAndIncrement();

sink.next(i * i); //2

if (i == 10) sink.complete(); //3

return state;

});

1.
The constructor of AtomicLong is used as the supplier.
2.
After incrementing, supply the square of the number to the sink.
3.
When the number is 10, the complete() method is called, which calls onComplete to any subscriber, closing out the Flux.

The create method takes a Consumer<? super FluxSink<T>> that exposes a FluxSink<T> instance with next, error, and complete methods. This allows you to arbitrarily publish data onto a Flux in any way you see fit.

The preceding code would produce a Flux of the squares of the numbers from zero to ten.

For example, the following demonstrates registering a MessageListener which handles a list of messages:

Flux<String> bridge = Flux.create(sink -> {

messageProcessor.register(

new MessageListener<String>() {

public void handle(List<String> chunks) {

for(String s : chunks) {

sink.next(s);

}

public void processComplete() {

sink.complete();

}

public void processError(Throwable e) {

sink.error(e);

}

});

Here sink’s type is FluxSink<String>. If the messages processed in the preceding code have a single-threaded source, the push method can be used instead of create . The push method has the same type signature as create, and so it is used in a similar way. FluxSink’s methods return FluxSink, allowing for method chaining, so the following example is possible:

Flux.push((FluxSink sink) -> {

sink.next(1).next(2).next(3).complete();

}).subscribe(System.out::println);

This would print out just the values 1, 2, and 3.

Schedulers

The Schedulers class under the reactor.core.scheduler package provides many static methods for Schedulers that determine what Thread or Threads your code will run on.

The following are some of those static methods and what they mean:

Schedulers.immediate() : The current thread.
Schedulers.single() : A single, reusable thread. Note that this method reuses the same thread for all callers, until the Scheduler is disposed. If you want a per-call dedicated thread, use Schedulers.newSingle() for each call.
Schedulers.newSingle() : Creates a new Thread each time it is called to be used by the underlying Flux.
Schedulers.elastic() : An elastic thread pool. It creates new worker pools as needed and reuses idle ones. Worker pools that stay idle for too long (default is 60 seconds) are disposed. This is a good choice for I/O blocking work for instance. Schedulers.elastic() is a handy way to give a blocking process its own thread, so that it does not tie up other resources.
Schedulers.parallel() : A fixed pool of workers. It creates as many workers as you have CPU cores.
Schedulers.fromExecutor(Executor) : Creates a Scheduler to use the given Executor, allowing you to use your extensive knowledge of Java’s Executors.

For example, let’s take our example of generating squares and make it run in parallel:

List<Integer> squares = new ArrayList<>();

Flux.range(1, 64).flatMap(v -> // 1

Mono.just(v)

.subscribeOn(Schedulers.newSingle("comp"))

.map(w -> w * w)) //2

.doOnError(ex -> ex.printStackTrace()) // 3

.doOnComplete(() -> System.out.println("Completed")) // 4

.subscribeOn(Schedulers.immediate())

.subscribe(squares::add); //5

1.
First we use Flux.range to take the range from 1 to 64 and call flatMap (which takes a lambda expression that converts each value in the range into a new Reactor type, Mono in this case).
2.
Using Schedulers.newSingle(name), we create a new single thread for each value, and passing to subscribeOn will cause the mapping expression to be executed on that single thread. Keep in mind we are describing the execution of the Mono here, not the initial Flux.
3.
We provide exception handling code using doOnError just in case.
4.
Using doOnComplete we print out “Completed” when the whole execution is finished.
5.
Finally, we subscribe to the Flux (without this step, nothing would ever happen) and add the result to our list of squares.

The result of running this code will be that the squares List has the value of every square from 1 to 64.

Here we see once again how in Reactive Streams everything can become a stream, even a single value. By creating a Mono for each value in the range, we’re able to use Reactor to declare what kind of threading we want for every calculation. In this case, since we are using newSingle , all of the processing will be done in parallel with a new thread for all 64 values.

However, this is probably not the most efficient implementation since creating lots of Threads causes a lot of overhead. Instead, we should use Schedulers.parallel() so that the exact number of Threads your CPU can handle will be created. In this way, Reactor takes care of the details for you.

Pull Events

If you have more of a “pull” situation (events are created by polling a source), you can use the Flux.create(FluxSink) method. For example, the following code creates a Flux that polls a channel (some imaginary class representing a stream of Strings from outside of Reactor) for new events:

Flux<String> bridge = Flux.create(sink -> {

sink.onRequest(n -> channel.poll(n)) //1

.onCancel(channel::cancel) // 2

.onDispose(channel::close); // 3

channel.register(sink::next); //4

});

1.
Poll for events from the channel when requests are made with the given number. This “n” is the number of items requested.
2.
Call the channel’s cancel method when the Flux is cancelled.
3.
The channel.close method is given to onDispose to be invoked for complete, error, or cancel.
4.
Finally, register the sink’s “next” method as a listener to the channel.

Keep in mind that the Consumer passed to onRequest will not be called multiple times for no reason. It will be called with some number (e.g., 256) and then not called again until a significant number of items have been published to the Flux (i.e., sink.next called many times).

Reminder The code examples used in this book are available on GitHub .

Handling Backpressure

Reactor, like all implementations of Reactive Streams, has the ability to handle backpressure. Simply use one of the following methods on a Flux (or others not listed) to specify which backpressure strategy you want to use:

onBackpressureBuffer() : Buffers all items until they can be handled downstream.
onBackpressureBuffer(maxSize) : Buffers items up to the given count.
onBackpressureBuffer(maxSize, BufferOverflowStrategy) : Buffers items up to the given count and allows you to specify the strategy to use when and if the buffer is full. BufferOverflowStrategy is an enum that has three values: DROP_OLDEST, which drops the oldest items in the buffer, DROP_LATEST which drops the newer items, and ERROR which would terminate the stream with an error.
onBackpressureLatest() : Similar to keeping a buffer of only the last item added. If the downstream does not keep up with upstream, only the latest element will be given downstream.
onBackpressureError() : Ends the Flux with an error (calling the downstream Subscriber’s onError) with an IllegalStateException from Exceptions.failWithOverflow() if more items were produced upstream than requested downstream.
onBackpressureDrop() : Drops any items produced above what was requested. This would be useful, for example, in UI code to drop user input that can’t be handled immediately.
onBackpressureDrop(Consumer) : Drops any items produced above what was requested and calls the given Consumer for each dropped item.

With each of these methods, the strategy only applies when items are produced on the stream faster than they can be handled. If that’s not the case, for example, with a cold stream, no backpressure strategy is necessary.

For example, we might want to take a Flux named “bridge” created earlier that is a stream of user input and buffer up to 256 items like the following:

bridge.onBackpressureBuffer(256)

Also keep in mind that Reactor is not magic, and some care should be taken when considering backpressure strategies.

Reactor has excellent online documentation for consideration.

Context

Since version 3.1.0, Reactor comes with an advanced feature that is somewhat comparable to ThreadLocal but applied to a Flux or a Mono instead of a Thread: the Context.

Reactor’s Context is much like an immutable Map or key/value store. It is stored transparently from the Subscriber upward through the Subscription. Context is Reactor specific and does not work with the other Reactive Streams implementations.

When setting up the Context, you should not define it toward the beginning of the Flux. This is because it starts at the subscriber and is passed upstream. For example, do not do this:

// this is WRONG!

Flux<Integer> flux =

Flux.just(1).subscriberContext(Context.of("pid", 123));

Instead you should define it toward the end since it propagates “backward” up the chain. For example:

Flux<Integer> flux = Flux.just(1); //1

Flux<String> stringFlux = flux.flatMap(i ->

Mono.subscriberContext().map(ctx -> i + " pid: " +

ctx.getOrDefault("pid", 0))); //2

// supply context here:

StepVerifier.create( //3

stringFlux.subscriberContext(Context.of("pid", 123)))

.expectNext("1 pid: 123") //4

.verifyComplete();

1.
Create a Flux of just one value.
2.
Use flatMap, access the Context, and use it to create a String value using the “pid” key. We use the static method on Mono, subscriberContext() to access the value from the Context by calling “getOrDefault” on it.
3.
This uses the StepVerifier (which we cover next) to verify that we get the expected value. The StepVerifier subscribes to the Flux after setting the Context using the “subscriberContext” method.
4.
Call “expectNext” with a value of “1 pid: 123” which is what we expect from setting the value 123 with the key of “pid” on the Context.

Context is useful for storing data that is peripheral to the Flux, but still important. For example, sometimes we have some identifier that represents the action or the user that initiated an action, and we want to include it in log outputs (like what MDC is used for in logback).

Testing

Automated testing is always a good idea, and it would be nice to have tools to directly test Reactive Streams. Luckily, Reactor comes with a few elements dedicated to testing which are gathered into their own artifact we included earlier: reactor-test.

The two main uses of reactor-test are the following:

Testing that a sequence follows a given scenario with StepVerifier
Producing data in order to test the behavior of operators (including your own operators) downstream with TestPublisher

StepVerifier

Reactor’s StepVerifier can be used to verify the behavior of a Reactor Publisher (Flux or Mono). StepVerifier is an interface used for testing that can be created using one of several static methods on StepVerifier itself.

Here’s a simple example of a JUnit test utilizing StepVerifier:

@Test

public void testStepVerifier_Mono_error() {

Mono<String> monoError = Mono.error(

new RuntimeException("error")); //1

StepVerifier.create(monoError) //2

.expectErrorMessage("error") //3

.verify(); //4

}

1.
Create a Mono wrapping a RuntimeException imitating an actual error state.
2.
Create a StepVerifier wrapping that Mono.
3.
Declare that an onError event is expected and the Exception’s error message is “error”.
4.
Must call verify() at the end. This will throw an AssertionError if any expectations are not met.

We can also create a Mono of just one string and verify it, for example:

@Test public void testStepVerifier_Mono_foo() {

Mono<String> foo = Mono.just("foo"); //1

StepVerifier.create(foo) //2

.expectNext("foo") //3

.verifyComplete(); //4

}

1.
Create a Mono wrapping one value, “foo”.
2.
Create a StepVerifier wrapping that Mono.
3.
Expect onNext is called with “foo”.
4.
Call verifyComplete() has the same effect as verify() but also expects onComplete was called.

Here we will test a Flux with three values and timeout if it takes too long:

@Test public void testStepVerifier_Flux() {

Flux<Integer> flux = Flux.just(1, 4, 9); //1

StepVerifier.create(flux) //2

.expectNext(1) //3

.expectNext(4)

.expectNext(9)

.expectComplete() //4

.verify(Duration.ofSeconds(10)); //5

}

1.
Create a Flux of just three numbers.
2.
Create a StepVerifier wrapping that Flux.
3.
Call expectNext for each value expected.
4.
Call expectComplete to expect onComplete to be called.
5.
Finally, you must call verify() at the end. This variation of verify takes a Duration timeout value. Here it is 10 seconds. This can be useful to prevent the Test from hanging in cases where a Publisher might never call onComplete.

TestPublisher

The TestPublisher<T> class offers the ability to provide finely tuned data for test purposes. TestPublisher is a Reactive Streams Publisher<T> but can be converted to either a Flux or Mono using flux() or mono() methods.

TestPublisher has the following methods:

next(T) and next(T, T…): Triggers 1−n onNext signals.
emit(T…): Does the same as next and also terminates with an onComplete signal.
complete(): Terminates with an onComplete signal.
error(Throwable): Terminates with an onError signal.

The following demonstrates how you might use TestPublisher :

TestPublisher<Object> publisher = TestPublisher.create(); //1

Flux<Object> stringFlux = publisher.flux(); //2

List list = new ArrayList(); //3

stringFlux.subscribe(next -> list.add(next), ex ->

ex.printStackTrace()); //4

publisher.emit("foo", "bar"); //5

assertEquals(2, list.size()); //6

assertEquals("foo", list.get(0));

assertEquals("bar", list.get(1));

1.
Create the TestPublisher instance.
2.
Convert it to a Flux.
3.
Create a new List. For test purposes we will use this list to collect values from the publisher.
4.
Subscribe to the publisher using two lambda expressions for onNext and onError. This will add each value emitted from the publisher to the list.
5.
Emit the values “foo” and “bar” from the TestPublisher.
6.
Assert that two values were added to the list and they are what we expect.

Note that you must subscribe to the TestPublisher before emitting any values.

5. Reactor