Reactor 3 Reference Guide

The Reactor reference guide is available as html documents. The latest copy is available at http://projectreactor.io/docs/core/release/reference/docs/index.html

Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.

There are several ways to reach out for help with Reactor.

Reactor is a fully non-blocking reactive programming foundation for the JVM, with efficient demand management (backpressure). It integrates directly with Java 8 functional APIs, notably CompletableFuture, Stream and Duration. It offers composable asynchronous sequence APIs Flux ([N] elements) and Mono ([0|1] elements), extensively implementing the Reactive Extensions specification.

Reactor also supports non-blocking IPC with the reactor-ipc components. Suited for Microservices Architecture, Reactor IPC offers backpressure-ready network engines for HTTP (including Websockets), TCP and UDP. Reactive Encoding/ Decoding is fully supported.

Reactor 3 uses a BOM^[1] model since reactor-core 3.0.4, with the Aluminium release train.

This allows to regroup artifacts that are meant to work well together without having to wonder about the sometimes divergent versioning schemes of these artifacts.

The BOM is like a curated list of versions. It is itself versioned, using a release train scheme with a codename followed by a qualifier:

The codenames represent what would traditionally be the MAJOR.MINOR number. They come from the Periodic Table of Elements (mostly), in growing alphabetical order.

The qualifiers are (in chronological order):

As mentioned above, the easiest way to use Reactor in your core is to use the BOM and add the relevant dependencies to your project. Note that when adding such a dependency, you omit the version so that it gets picked up from the BOM.

However, if you want to force the use of a specific artifact’s version, you can specify it when adding your dependency, as you usually would. You can also of course forgo the BOM entirely and always specify dependencies with their artifact versions.

Modern applications nowadays can reach huge scales of users, and even though the capabilities of modern hardware have continued to improve, performance of the modern software is still a key concern.

There are broadly two ways one can improve a program’s performance:

Usually, Java developers will naturally write program using blocking code. This is all well until there is a performance bottleneck, at which point the time comes to introduce additional thread(s), running similar blocking code. But this scaling in resource utilization can quickly introduce contention and concurrency problems.

Worse! If you look closely, as soon as a program involves some latency (notably I/O, like a database request or a network call), there is a waste of resources in the sense that the thread now sits idle, waiting for some data.

So the parallelization approach is not a silver bullet: although it is necessary in order to access the full power of the hardware, it is also complex to reason about and susceptible to resource wasting…​

The second approach described above, seeking more efficiency, can be a solution to that last problem. By writing asynchronous non-blocking code, you allow for the execution to switch to another active task using the same underlying resources, and to later come back to the current "train of thought" when the asynchronous processing has completed.

But how can you produce asynchronous code on the JVM?

Java offers mainly two models of asynchronous programming:

So is it good enough? Well, not for every use cases, and both approaches have limitations…​

Callbacks are very hard to compose together, quickly leading to code that is difficult to read and maintain ("Callback Hell").

Futures are a bit better, but they are still not so good at composition, despite the improvements brought in Java 8 by CompletableFuture…​ Orchestrating multiple futures together is doable, but not that easy. Plus it is very (too?) easy to stay in familiar territory and block on a Future by calling their get() method. And lastly, they lack the support for multiple values and advanced error handling.

This might seem familiar: isn’t that what Reactive Programming directly tries to address with the Publisher-Subscriber pair?

Indeed, reactive libraries like Reactor aim at addressing these drawbacks of "classic" asynchronous approaches on the JVM, while also focusing on a few additional aspects. To sum it up:

By composability, we mean the ability to orchestrate multiple asynchronous tasks together, using results from previous tasks to feed input to subsequent ones, or executing several tasks in a fork-join style, as well as reusing asynchronous tasks as discrete components in an higher level system.

This is tightly coupled to readability and maintainability of one’s code, as these layers of asynchronous processes get more and more complex. As we saw, the callback model is simple, but one of its main drawbacks is that for complex processes you need to have a callback executed from a callback, itself nested inside another callback, and so on…​

That is what is referred to as Callback Hell. And as you can guess (or know from experience), such code is pretty hard to go back to and reason about.

Reactor on the other hand offers rich composition options where code mirrors the organization of the abstract process, and everything is kept at the same level (no nesting if it is not necessary).

You can think of data processed by a reactive application as moving through an assembly line. Reactor is the conveyor belt and working stations. So the raw material pours from a source (the original Publisher) and ends up as a finished product ready to be pushed to the consumer (or Subscriber).

It can go to various transformations and other intermediary steps, or be part of a larger assembly line that aggregates intermediate pieces together.

Finally, if there is a glitch or a clogging at one point (for example boxing the products takes a disproportionately long time), the workstation can signal that upstream and limit the flow of raw material.

In Reactor, operators are what we represented in the above analogy as the assembly line’s workstations. Each operator adds behavior to a Publisher, and it actually wraps the previous step’s Publisher into a new instance.

The whole chain is thus layered, like an onion, where data originates from the first Publisher in the center and moves outward, transformed by each layer.

While the Reactive Streams specification doesn’t specify operators at all, one of the high added values of derived reactive libraries like Reactor is the rich vocabulary of operators that they bring along. These cover a lot of ground, from simple transformation and filtering to complex orchestration and error handling.

In Reactor when you write a Publisher chain, data doesn’t start pumping into it by default. Instead, what you have is a abstract description of your asynchronous process (which can help with reusability and composition by the way).

By the act of subscribing, you tie the Publisher to a Subscriber, which triggers the flow of data in the whole chain. This is achieved internally by a single request signal from the Subscriber that is propagated upstream, right back to the source Publisher.

The same mechanism is in fact used to implement backpressure, which we described in the assembly line analogy as a feedback signal sent up the line when a working station is slower to process than the upstream.

The real mechanism defined by the Reactive Streams specification is pretty close to the analogy: a subscriber can work in unbounded mode and let the source push all the data at its fastest achievable rate, but can also use the request mechanism to signal the source that it is ready to process at most n elements.

Intermediate operators can also change the request in-flight. Imagine a buffer operator that groups elements in batches of 10. If the subscriber requests 1 buffer, then it is acceptable for the source to produce 10 elements. Prefetching strategies can also be applied is producing the elements before they are requested is not too costly.

This transforms the push model into a push-pull hybrid where the downstream can pull n elements from upstream if they are readily available, but if they’re not then they will get pushed by the upstream whenever they are produced.

In the Rx family of reactive libraries, one can distinguish two broad categories of reactive sequences: hot and cold. This distinction mainly has to do with how the reactive stream reacts to subscribers:

For more information on hot vs cold in the context of Reactor, see this reactor-specific section.

A Flux<T> is a standard Publisher<T> representing an asynchronous sequence of 0 to N emitted items, optionally terminated by either a success signal or an error.

As in the RS spec, these 3 types of signal translate to calls to downstream’s onNext, onComplete or onError methods.

With this large scope of possible signal, Flux is the general-purpose reactive type. Note that all events, even terminating ones, are optional: no onNext event but an onComplete event represents an empty finite sequence, but remove the onComplete and you have an infinite empty sequence. Similarly, infinite sequences are not necessarily empty: Flux.interval(Duration) produces a Flux<Long> that is infinite and emits regular ticks from a clock.

A Mono<T> is a specialized Publisher<T> that emits at most one item then optionally terminates with an onComplete signal or an onError.

As such it offers only a relevant subset of operators. For instance, combination operators can either ignore the right hand-side emissions and return another Mono or emit values from both sides, in which case they’ll switch to a Flux.

Note that a Mono can be used to represent no-value asynchronous processes that only have the concept of completion (think Runnable): just use an empty Mono<Void>.

The easiest way to get started with Flux and Mono is to use one of the numerous factory methods found in their respective classes.

For instance, to create a simple sequence of String, you can either enumerate them or put them in a collection and create the Flux from it:

Other examples of factory methods include:

When it comes to subscribing, Flux and Mono make use of Java 8 lambdas. You have a wide choice of .subscribe() variants that take lambdas for different combinations of callbacks:

These are convenience variant over the Reactive Streams defined subscribe:

That last variant is useful if you already have a Subscriber handy, but more often you’ll need it because you want to do something subscription-related in the other callbacks. Most probably, that’d be dealing with backpressure and triggering the requests yourself.

In that case, you can ease things up by using the BaseSubscriber abstract class, which offers convenience methods for that:

BaseSubscriber also offers a requestUnbounded() method to switch to unbounded mode (equivalent to request(Long.MAX_VALUE).

In this section, we’ll introduce means of creating a Flux (or Mono) by programmatically defining its associated events (onNext, onError, onComplete). All these methods share the fact that they expose an API to trigger the events that we call a sink. There are actually a few sink variants, as you will discover below.

Reactor, like RxJava, can be considered concurrency agnostic. It doesn’t enforce a concurrency model but rather leave you, the developer, in command.

But that doesn’t prevent the library from helping you with concurrency…​

In Reactor, the execution model and where the execution happens is determined by the Scheduler that is used. A Scheduler is an interface that can abstract a wide range of implementations. The Schedulers class has static methods that give access to the following execution contexts:

Additionally, you can create a Scheduler out of any pre-existing ExecutorService ^[4] using Schedulers.fromExecutorService(ExecutorService), and also create new instances of the various scheduler types using newXXX methods.

Some operators use a specific Scheduler from Schedulers by default (and will usually give you the option of providing a different one). For instance, calling the factory method Flux.interval(Duration.ofMillis(300)) will produces a Flux<Long> that ticks every 300ms. This is enabled by Schedulers.timer() by default.

Reactor offers two means of switching execution context (or Scheduler) in a reactive chain: publishOn and subscribeOn. Both take a Scheduler and allow to switch the execution context to that scheduler. But publishOn placement in the chain matters, while subscribeOn's doesn’t. To understand that difference, you first have to remember that Nothing happens until you subscribe().

In Reactor, when you chain operators you wrap as many Flux/Mono specific implementations inside one another. And as soon as you subscribe, a chain of Subscriber is created backward. This is effectively hidden from you and all you can see is the outer layer of Flux (or Mono) and Subscription, but these intermediate operator-specific subscribers are where the real work happens.

With that knowledge, let’s have a closer look at the two operators:

In Reactive Streams, errors are terminal events. As soon as an error occurs, it stop the sequence and gets propagated down the chain of operators to the last step, the Subscriber you defined and its onError method.

Such errors should still be dealt with at the application level, for instance by displaying an error notification in a UI, or sending a meaningful error payload in a REST endpoint, so the subscriber’s onError method should always be defined.

But Reactor also offers alternative means of dealing with errors in the middle of the chain, as error-handling operators.

Let’s go through each mean of error handling one-by-one. When relevant we’ll make a parallel with imperative world’s try patterns.

Processors are a special kind of Publisher that are also a Subscriber. That means that you can subscribe to a Processor (generally, they implement Flux), but also call methods to manually inject data into the sequence or terminate it…​

There are several kind of Processors, each with a few particular semantics, but before you start looking into these, you need to ask yourself the following question:

The most common case for testing a Reactor sequence is to have a Flux or Mono defined in your code (eg. returned by a method), and wanting to test how it behaves when subscribed to.

This translates well to defining a "test scenario", where you define your expectations in terms of events, step-by-step: what is the next expected even? Do you expect the Flux to emit a particular value? Or maybe to do nothing for the next 300ms? All of that can be expressed through the StepVerifier API.

For instance, you could have the following utility method in your codebase that decorates a Flux:

So in order to test it, you’d want to verify the following scenario:

In the StepVerifier API, this translates to:

The API is a builder. You start by creating a StepVerifier and passing the sequence to be tested. This offers a choice of methods that allow you to:

For terminal events, the corresponding expectation methods (expectComplete(), expectError() and all its variants) will switch to an API where you cannot express expectations anymore. In that last step, all you can do is perform some additional configuration on the StepVerifier then trigger the verification.

What happens at this point is that the StepVerifier subscribes to the tested flux/mono and plays the sequence, comparing each new signal with the next step in the scenario. As long as these match, the test is considered a success. As soon as there is a discrepancy, an AssertionError is thrown.

Note that if one of the lambda-based expectations throws an AssertionError, it will be reported as is, failing the test. This is useful for custom assertions.

Another very interesting capability of StepVerifier is the way it can be used with time-based operators in order to avoid long run times for corresponding tests. This is done through the StepVerifier.withVirtualTime builder.

It looks like this:

The way this virtual time feature works is that it plugs in a custom Scheduler in Reactor’s Schedulers factory. Since these timed operators usually use the default Schedulers.timer() scheduler, replacing it with a VirtualTimeScheduler does the trick. However, an important pre-requisite is that the operator be instantiated after the virtual time scheduler has been activated.

In order to increase the chances this happens correctly, the StepVerifier won’t take a simple Flux as input. withVirtualTime takes a Supplier, which allows to lazily create the instance of the tested flux AFTER having done the scheduler set up.

There are a couple of expectation methods that deal with time, and they are both valid with or without virtual time:

Both methods will pause the thread for the given duration in classic mode, and advance the virtual clock instead in virtual mode.

So in order to quickly evaluate the behavior of our Mono.delay above, we can finish writing up our code like this:

We could have used thenAwait(Duration.ofDays(1)) above, but expectNoEvent has the benefit of guaranteeing that nothing happened earlier that it should have.

Note also that verify() returns a Duration value. This is the real time duration of the entire test.

After having described the final expectation of your scenario, you can switch to a complementary assertion API instead of plainly triggering the verify(): use verifyThenAssertThat() instead.

This returns a StepVerifier.Assertions object which you can use to assert a few elements of state once the whole scenario has played out successfully (since it does also call verify()). Typical (albeit advanced) usage is to capture elements that have been dropped by some operator and assert them (see the section on Hooks).

For more advanced test cases, it might be useful to have complete mastery over the source of data, in order to trigger finely chosen signals that closely match the particular situation you want to test.

Another situation is when you have implemented your own operator and you want to verify how it behaves with regards to the Reactive Streams specification, especially if its source is not well behaved.

For both cases, reactor-test offers the TestPublisher. This is a Publisher<T> that lets you programmatically trigger various signals:

A well-behaved TestPublisher can be obtained through the create factory method. Additionally, misbehaving TestPublisher can be created using the createNonCompliant factory method. The later takes a number of Violation enums that will define which parts of the specification the publisher can overlook. For instance:

Finally, the TestPublisher keeps track of internal state after subscription, which can be asserted through its various assertXXX methods.

It can be used as a Flux or Mono by using the conversion methods flux() and mono().

But as soon as you shift to asynchronous code, things can get much more complicated…​

Consider the following stacktrace:

There is a lot going on there! We get an IndexOutOfBoundsException which tell us that a "source emitted more than one item".

We can probably quickly come to assume that this source is a Flux/Mono, as confirmed by the line below that mentions MonoSingle. So it appears to be some sort of complaint from a single operator.

Referring to the javadoc for Mono#single operator, we indeed remember that single has a contract: the source must emit exactly one element. It appears we had a source that emitted more than one and thus violated that contract.

Can we dig deeper and identify that source? The following rows don’t seem very helpful. They take us on a travel inside the internals of what seems to be a reactive chain, through subscribes and requests…​

By skimming over these rows, we can at least start to form a picture of the kind of chain that went wrong: it seems to involve a MonoSingle, a FluxFlatMap and a FluxRange (each get several rows in the trace, but overall these 3 classes are involved). So a range().flatMap().single() chain maybe?

But what if we use that pattern a lot in our application? This still doesn’t tell us much, and simply searching for single isn’t going to cut it. Then the last line refers to some of our code. Finally!

Hold on…​ When we go to the source file, all we see is that a pre-existing Flux is subscribed to:

All of this happened at subscription time, but the Flux itself wasn’t declared there. Worse, when we go to where the variable is declared, we see:

The variable isn’t even instantiated where it is declared. Let’s assume a worst case scenario where we find out there could be a few different codepath that set it in the application…​ So we’re still unsure of which one caused the issue.

What we want to find out more easily is where the operator was added into the chain, where the Flux was declared. We usually refer to that as the assembly of the Flux.

Even though the stacktrace was still able to convey some information for someone with a bit of experience, we can see that it is not ideal by itself in more advanced cases.

Fortunately, Reactor comes with a debugging-oriented capability of assembly-time instrumentation.

This is done by customizing the Hook.onOperator hook at application start (or at least before the incriminated flux or mono can be instantiated), like so:

The idea is that this will start instrumenting the calls to Flux (and Mono)'s operator methods (where they are assembled into the chain) by wrapping the construction of the operator and capturing a stacktrace there. Since this is done when the operator chain is declared, the hook should be activate before that, so the safest way is to activate it right at the start of your application.

Later on, if an exception occurs, the failing operator will be able to refer to that capture and append it to the stacktrace.

In the next section, we’ll see how the stacktrace differs and how to interpret that new information.

Reusing our initial example but activating the operatorStacktrace debug feature, here is the stack we now get:

As you can see, the captured stacktrace is appended to the original error as a suppressed OnAssemblyException. There are two parts to it, but the first section is the most interesting. It shows the path of construction for the operator that triggered the exception. Here it shows that the single that caused our issue was created in the scatterAndGather method, itself called from a populateDebug method that got executed through JUnit.

We are now armed with enough information to find the culprit, let’s have a look at that scatterAndGather method:

Now we can see what the root cause of the error was: a flatMap that performs several HTTP calls to a few urls is chained with single, which seem a bit too restrictive. After a short git blame and a quick discussion with the author of that line, we find out he meant to use the less restrictive take(1) instead…​

Congratulations, we solved our problem!

That second part of the debug stacktrace was not necessarily very interesting in this particular example, because the error was actually happening in the last operator in the chain (the one closest to subscribe). Taking another example might make it clearer:

Now imagine that inside findAllUserByName there is a map that fails. Here we would see the following final traceback:

This correspond to a flattened out version of the chain of operators, or rather of the section of the chain that gets notified of the error:

Additionally to stacktrace debugging and analysis, another powerful tool to have in your toolbelt is the capability to trace and log events in an asynchronous sequence.

The log() operator can do just that. Chained inside a sequence, it will peek at every event of the Flux/Mono upstream of it (including onNext, onError and onComplete of course, but also subscriptions, cancellation and requests).

The operator picks up common logging frameworks like Log4J and Logback through SLF4J, and will default to the JDK Logger in case none can be found.

For instance, supposing we have logback activated and configured, and a chain like range(1,10).take(3). By placing a log() just before the take, we can get some insight as to how it works and what kind of events it propagates upstream to the range:

This prints out (through the logger’s console appender):

Here, additionally to the logger’s own formatter (time, thread, level, message), the log() operator outputs a few things in its own format:

The last line (4) is the most interesting: we can see the take in action there: it operates by cutting the sequence short after it has seen enough elements emitted. In a word, take simply cancel() the source once it has emitted the user-requested amount!

From a clean code perspective, code reuse is generally a good thing. Reactor offers a few patterns that will help you reuse and mutualize code, notably for operators or combination of operators that you might want to apply regularly in your codebase.

So far we have considered that all Flux (and Mono) are the same: they all represent an asynchronous sequence of data, and nothing happens before you subscribe.

There are however in reality two broad families of publishers: cold ones and hot ones.

The description above applies to the cold family of publishers. They generate data anew for each subscription, and if no subscription is done then data never start generating.

Think HTTP request: each new subscriber will trigger an HTTP call, but no call is made if no one is interested in the result.

Hot publishers on the other hand don’t really depend on any number of subscribers. They might start publishing data right away, and would continue doing so whenever a new Subscriber comes in (in which case said subscriber would only see new elements emitted after it subscribed). So for such hot publishers, something indeed happens before you subscribe.

One example of the few hot operators in Reactor is just: it directly capture the value at assembly time, and will replay it to anybody subscribing to it later on. To re-use the HTTP call analogy, if the captured data is the result of an HTTP call then only one network call is made, when instantiating just.

To transform just into a cold publisher, you can use defer. This will defer the HTTP request in our example to subscription time (and would result in a separate network call for each new subscription).

Contrast these two other examples:

Which outputs:

Compared to:

Which outputs:

Sometimes, you don’t only want to defer some processing to the subscription time of one subscriber, but you might actually want for several of them to rendez-vous and then trigger the subscription / data generation.

This is what ConnectableFlux is made for. Two main patterns are covered in the Flux API that return a ConnectableFlux: publish and replay.

A ConnectableFlux offers additional methods to manage subscriptions downstream vs subscription to the original source. For instance:

This outputs:

With autoConnect:

Which outputs:

With multi-core architectures being a commodity nowadays, being able to easily parallelize work is very important. Reactor helps with that by providing a special type, ParallelFlux, that exposes operators that are optimized for parallelized work.

To obtain a ParallelFlux, one can use the parallel() operator on any Flux. This will not by itself parallelize the work however, but rather will divide the workload into "rails" (by default as many rails as there are CPU cores).

In order to tell the resulting ParallelFlux where to execute each rail (and by extension to execute rails in parallel) you have to use runOn(Scheduler). Note that there is a recommended dedicated Scheduler for parallel work: Schedulers.parallel().

Contrast:

with:

The first outputs:

While the second correctly parallelizes on two threads:

If once you’ve processed your sequence in parallel you want to revert back to a "normal" flux and apply the rest of the operator chain in a sequential manner, you can use the sequential() method on ParallelFlux.

Note that it is the case by default if you subscribe to the ParallelFlux with a single provided Subscriber, but not when using the lambda-based variants of subscribe.

You can also access individual rails or "groups" as a Flux<GroupedFlux<T>> via the groups() method and apply additional operators to them via the composeGroup() method.

Check you have affected the result of the operator to the variable you .subscribe() to.

Reactor operators are decorators, they return a different instance that wraps the source sequence and add behavior. That is why the preferred way of using operators is to chain the calls.

Compare the following:

With:

And even better:

The first version will output:

Whereas the two other versions will output the expected:

If the source Mono is either empty or a Mono<Void> (a Mono<Void> is empty for all intent and purposes), some combinations will never be called. You should expect that if there is a callback and the combination depends on the value…​

This includes the Function based variant of then (then(v -> Mono.just(1))) and all versions of and.

In order to aVoid this problem and still invoke the continuation lazily, when the source Mono terminates, use the Supplier based version:

The retryWhen operator can be quite complex. Hopefully this snippet of code can help you understand how it works by attempting to emulate a simpler retry(3):

Exponential backoff produces retry attempts with a growing delay between each of the attempts, so as not to overload the source systems and risk an all out crash. The rationale is that if the source errors, it is already in an unstable state, and not likely to immediately recover from it. So blindly retrying immediately is likely to produce yet another error and add to the instability.

Here is how to implement an exponential backoff that delays retries and increase the delay between each attempt (delay == attempt number * 100 milliseconds):

When subscribed to, this fails and terminates after printing out:

TODO

TODO