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.

The reference guide is written in Asciidoc format and sources can be found at https://github.com/reactor/reactor-core/tree/master/src/docs/asciidoc.

If you have an improvement, we will be happy to get a pull-request from you!

We recommend that you check out a local copy of the repository, so that you can generate the documentation using the asciidoctor gradle task and check the rendering. Some of the sections rely on included files, so GitHub rendering is not always complete.

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

Suggest Edit to "About the Documentation"

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

Reactor also supports non-blocking inter-process communication (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 and Decoding are fully supported.

Reactor Core runs on Java 8 and above.

It has a transitive dependency to org.reactive-streams:reactive-streams:1.0.1.

Reactor 3 uses a BOM (Bill of Materials - a standard Maven artifact) model since reactor-core 3.0.4, with the Aluminium release train.

The BOM lets you group artifacts that are meant to work well together without having to wonder about the sometimes divergent versioning schemes of these artifacts.

The BOM is a list of versioned artifacts that is itself versioned, using a release train scheme with a codename followed by a qualifier. Here is an example:

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

The qualifiers are (in chronological order):

As mentioned earlier, 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 must omit the version so that the version 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 forgo the BOM entirely and specify dependencies by their artifact versions.

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

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

Usually, Java developers write programs using blocking code. This practice is fine until there is a performance bottleneck, at which point the time comes to introduce additional threads, running similar blocking code. But this scaling in resource utilization can quickly introduce contention and concurrency problems.

Worse still, blocking wastes resources. If you look closely, as soon as a program involves some latency (notably I/O, such as a database request or a network call), resources are wasted because a thread (or many threads) now sits idle, waiting for data.

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

The second approach mentioned above, seeking more efficiency, can be a solution to the resource wasting problem. By writing asynchronous, non-blocking code, you let the execution switch to another active task using the same underlying resources and later come back to the current process when the asynchronous processing has finished.

But how can you produce asynchronous code on the JVM? Java offers two models of asynchronous programming:

Are these techniques good enough? Not for every use case, and both approaches have limitations.

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

Consider an example: showing top five favorites from a user on the UI or suggestions if she doesn’t have a favorite. This goes through three services (one gives favorite IDs, the second fetches favorite details, while the third offers suggestions with details):

That is a lot of code, and it is a bit hard to follow and has repetitive parts. Consider its equivalent in Reactor:

What if you wanted to ensure the favorite IDs are retrieved in less than 800ms or, if it takes longer, get them from a cache? In the callback-based code, that is a complicated task. But in Reactor it becomes as easy as adding a timeout operator in the chain:

Futures are a bit better, but they are still not good at composition, despite the improvements brought in Java 8 by CompletableFuture. Orchestrating multiple futures together is doable but not easy. Also, Future has other problems: It is easy to end up with another blocking situation with Future objects by calling the get() method, and they lack support for multiple values and advanced error handling.

Consider another example: We get a list of IDs from which we want to fetch a name and a statistic and combine these pair-wise, all of it asynchronously.

Since Reactor has more combination operators out of the box, this process can be simplified:

These perils of Callback and Future are similar and are what Reactive Programming addresses with the Publisher-Subscriber pair.

Reactive libraries such as Reactor aim to address these drawbacks of "classic" asynchronous approaches on the JVM while also focusing on a few additional aspects:

A Flux<T> is a standard Publisher<T> representing an asynchronous sequence of 0 to N emitted items, optionally terminated by either a completion signal or an error. Thus, the possible values of a flux are a value, a completion signal, or an error. As in the Reactive Streams spec, these 3 types of signal translate to calls to a downstream object’s onNext, onComplete or onError methods.

With this large scope of possible signals, 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. For example, 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 subset of the operators that are available for a Flux. 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.

For example, Mono#concatWith(Publisher) returns a Flux while Mono#then(Mono) returns another Mono.

Note that a Mono can be used to represent no-value asynchronous processes that only have the concept of completion (such as Runnable). To create one, 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 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:

In this section, we introduce the creation of a Flux or a Mono by programmatically defining its associated events (onNext, onError, and 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, which we’ll get to shortly.

Reactor, like RxJava, can be considered concurrency agnostic. That is, it does not enforce a concurrency model but rather leaves you, the developer, in command. However, 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 using Schedulers.fromExecutorService(ExecutorService). (You can also create one from an Executor, although doing so is discouraged.) You can also create new instances of the various scheduler types using newXXX methods. For example, Schedulers.newElastic(yourScheduleName) creates a new elastic scheduler named yourScheduleName.

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

Reactor offers two means of switching the execution context (or Scheduler) in a reactive chain: publishOn and subscribeOn. Both take a Scheduler and let you switch the execution context to that scheduler. But the placement of publishOn in the chain matters, while the placement of subscribeOn does not. To understand that difference, you first have to remember that nothing happens until you subscribe().

In Reactor, when you chain operators, you can wrap as many Flux and Mono implementations inside one another as you need. Once you subscribe, a chain of Subscriber objects is created, backward (up the chain) to the first publisher. This is effectively hidden from you. 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, we can have a closer look at the publishOn and subscribeOn operators:

Flux and Mono do not create threads. Some operators, such as publishOn, do create threads. Also, as a form of work sharing, these operators can "steal" threads from other work pools if the other pools are currently idle. Consequently, neither the Flux or Mono objects nor the Subscriber objects have to be smart about threads. They rely on the operators to manage the threads and work pools.

publishOn forces the next operator (and possibly subsequent operators after the next one) to run on a different thread. Similarly, subscribeOn forces the previous operator (and possibly operators prior to the previous one) to run on a different thread. Remember that, until you subscribe, you’re defining a process but not starting it. For that reason, Reactor can use these rules to determine how the processing must proceed. Then, when you subscribe, the whole process starts working.

Consider an example showing that multiple threads can support work sharing:

Scheduler.parallel() creates a fixed pool of single-threaded ExecutorService-based workers. Because one or two threads may lead to problems, it always creates at least four threads. The publishOn method then shares these threaded workers, getting items from whichever is emitting when publishOn requests an item. In this fashion, we get work sharing (a form of resource sharing). Reactor enables several other ways to share for resources as well, as documented in Schedulers.

Scheduler.elastic() also creates threads and is a handy way to create a dedicated thread in the case of wrapping a blocking resource (such as a synchronous service). See How do I wrap a synchronous, blocking call?.

Behind the scenes, the operators ensure thread safety by using incremental counters and guard conditions. For instance, if we have four threads hitting a single stream (as shown in the preceding example) each request increments a counter, so that future requests from the threads get the right item.

In Reactive Streams, errors are terminal events. As soon as an error occurs, it stops 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, you might display an error notification in a UI or send a meaningful error payload in a REST endpoint. For this reason, the subscriber’s onError method should always be defined.

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

Now we can consider each means of error handling one-by-one. When relevant, we make a parallel with imperative programming’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 you can 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:

Suggest Edit to "Which operator do I need?"

The most common case for testing a Reactor sequence is to have a Flux or Mono defined in your code (for example, 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 event? 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 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() and expectError() and all its variants) 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 and then trigger the verification, eg. with verify().

What happens at this point is that the StepVerifier subscribes to the tested Flux or 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 is 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.parallel() 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 does not take a simple Flux as input. withVirtualTime takes a Supplier, which allows for lazily creating the instance of the tested flux AFTER having done the scheduler set up.

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

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

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

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

Note 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 triggering verify(). To do so, use verifyThenAssertThat() instead.

This method 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 also calls 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 class. This is a Publisher<T> that lets you programmatically trigger various signals:

A well behaved TestPublisher can be obtained through the create factory method. Also, a misbehaving TestPublisher can be created using the createNonCompliant factory method. The latter takes a number of Violation enums that 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 assert* methods.

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

Suggest Edit to "Testing"

When you shift to asynchronous code, things can get much more complicated.

Consider the following stack trace:

There is a lot going on there. We get an IndexOutOfBoundsException, which tells 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 see 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 find the problem. Then the last line refers to some of our code. Finally, we are getting close.

Hold on, though. 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 was not declared there. Worse, when we go to where the variable is declared, we see:

The variable is not instantiated where it is declared. We must assume a worst-case scenario where we find out that there could be a few different code paths that set it in the application. 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 starts instrumenting the calls to the Flux (and Mono) 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 activated 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 is able to refer to that capture and append it to the stack trace.

In the next section, we see how the stack trace differs and how to interpret that new information.

Reusing our initial example but activating the operatorStacktrace debug feature, the stack trace is now this:

As you can see, the captured stack trace 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.

Now that we are armed with enough information to find the culprit, we can have a meaningful 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 is 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.

We have solved our problem.

That second part of the debug stack trace was not necessarily interesting in this particular example, because the error was actually happening in the last operator in the chain (the one closest to subscribe). Considering another example might make it more clear:

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

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

We are dealing with a form of instrumentation here, and creating a stack trace is costly. That is why this debugging feature should only be activated in a controlled manner, as a last resort.

In addition to stack trace debugging and analysis, another powerful tool to have in your toolkit is the ability 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 but also subscriptions, cancellations and requests).

For instance, suppose 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, in addition 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 short, take() causes the source to cancel() once it has emitted the user-requested amount.

Suggest Edit to "Debugging Reactor"

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. If you think of a chain of operators as a recipe, you can create a cookbook of operator recipes.

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.

Really, though, there are two broad families of publishers: cold and hot.

The description above applies to the cold family of publishers. They generate data anew for each subscription If no subscription is created, then data never gets generated.

Think of an 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 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). For hot publishers, something does indeed happen before you subscribe.

One example of the few hot operators in Reactor is just: it directly captures 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. It defers 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:

This first example produces:

Both subscribers catch all four colors, because each subscriber causes the process defined by the operators on the Flux to run.

Compare the first example to this second example:

The second example produces:

Subscriber 1 catches all four colors. Subscriber 2, having been created after the first two colors were produced, catches only the last two colors. This difference accounts for the doubling of "ORANGE" and "PURPLE" in the output. The process described by the operators on this Flux runs regardless of when subscriptions have been attached.

Sometimes, you want to not only defer some processing to the subscription time of one subscriber, but you might actually want for several of them to rendezvous 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:

Consider the following example:

This code produces:

With autoConnect:

Which outputs:

When you have lots of elements and you want to separate them into batches, you have three broad solutions in Reactor: grouping, windowing, and buffering. These three are conceptually close, because they redistribute a Flux<T> into an aggregate. Grouping and windowing create a Flux<Flux<T>>, while buffering aggregates into Collection<T>.

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().

Compare the next two code examples:

with:

The first code block produces:

The second correctly parallelizes on two threads, as shown here:

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 sequential() is implicitly applied if you subscribe to the ParallelFlux with a Subscriber, but not when using the lambda-based variants of subscribe.

Note also that subscribe(Subscriber<T>) merges all the rails, while subscribe(Consumer<T>) runs all the rails. If the subscribe() method has a lambda, each lambda is executed as many times as there are rails.

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.

Suggest Edit to "Advanced features and concepts"

It’s often the case that a source of information is synchronous and blocking. To deal with such sources in your Reactor applications, apply the following pattern:

You should use a Mono because the source returns one value. You should use Schedulers.elastic because it creates a dedicated thread to wait for the blocking resource without tying up some other resource.

Note that subscribeOn does not subscribe to the Mono. It specifies what kind of Scheduler to use when a subscribe call happens.

Make sure that the variable you .subscribe() to has been affected by the operators you think should have been applied to it.

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 two examples:

This sample is even better (because it’s simpler):

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 intents and purposes), some combinations will never be called.

This is the typical case for any transformer like the zipWith operator, which by definition needs an element from each source to produce its output. If any of the sources are empty, and produces an empty sequence as well, so be careful of its usage. For example, using zipWith() after a then() may cause this problem.

This is even more true of the variants of and that take a Function, meaning that the associated Mono is chosen lazily depending on the incoming value (which never comes in the cases of empty or Void sequences).

You can use .defaultIfEmpty(T) to replace an empty sequence of T (not a Void sequence) with a default value, which could help avoid some of these situations. Here’s an example this:

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 produces an error, 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 (pseudocode: delay = attempt number * 100 milliseconds):

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

As described in Schedulers, publishOn() can be used to switch execution contexts. The publishOn operator influences the threading context where the rest of the operators in the chain below it will execute, up to a new occurrence of publishOn. So the placement of publishOn is significant.

For instance, in the example below, the transform function in map() is executed on a worker of scheduler1 and the processNext method in doOnNext() is executed on a worker of scheduler2. Single threaded schedulers may be used to ensure thread affinity for different stages in the chain or for different subscribers.

Suggest Edit to "FAQ, Best Practices, and "How do I…​?""

TODO

TODO