Reactor 3 Reference Guide

The Reactor reference guide is available as HTML documents. The latest copy is available at https://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/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 Streams](https://www.reactive-streams.org/) specification.

Reactor also supports non-blocking inter-process communication with the reactor-netty project. Suited for Microservices Architecture, Reactor Netty 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 on org.reactivestreams:reactive-streams:1.0.2.

Reactor 3 uses a BOM model since reactor-core 3.0.4, with the Aluminium release train. This curated list groups artifacts that are meant to work well together, providing the relevant versions despite potentially divergent versioning schemes in these artifacts.

The BOM (Bill of Materials) is itself versioned, using a release train scheme with a codename followed by a qualifier. Here are a few examples:

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 earlier), 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 the 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, and 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 want 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. In Reactor it becomes as easy as adding a timeout operator in the chain:

Futures are a bit better than callbacks, but they still do not do well 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, they do not support lazy computation 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. As in the Reactive Streams spec, these 3 types of signal translate to calls to a downstream Subscriber’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 (not particularly useful, except for tests around cancellation). 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 and then optionally terminates with an onComplete signal or an onError signal.

It offers only a subset of the operators that are available for a Flux, and some operators (notably those that combine the Mono with another Publisher) 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 (similar to a 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, as follows:

Other examples of factory methods include the following:

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, as shown in the following method signatures:

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. Rather it leaves you, the developer, in command. However, that does not prevent the library from helping you with concurrency.

Obtaining a Flux or a Mono doesn’t necessarily mean it will run in a dedicated Thread. Instead, most operators continue working in the Thread on which the previous operator executed. Unless specified, the topmost operator (the source) itself runs on the Thread in which the subscribe() call was made.

The code above produces the following output:

In Reactor, the execution model and where the execution happens is determined by the Scheduler that is used. A Scheduler has scheduling responsibilities similar to an ExecutorService, but having a dedicated abstraction allows to do more, notably acting as a clock and enabling a wider range of implementations (virtual time for tests, trampolining or immediate scheduling, etc…​).

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 by 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 by using the 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. By default, this is enabled by Schedulers.parallel(). The following line changes the Scheduler to a new instance similar to Schedulers.single():

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:

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. Here is an example:

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 kinds of Processors, each with a few particular semantics, but before you start looking into these, you need to ask yourself the following question:

Kotlin is a statically-typed language targeting the JVM (and other platforms) which allows writing concise and elegant code while providing a very good interoperability with existing libraries written in Java.

Reactor 3.1 introduces first-class support for Kotlin which is described in this section.

Reactor supports Kotlin 1.1+ and requires kotlin-stdlib (or one of its kotlin-stdlib-jre7 / kotlin-stdlib-jre8 variants).

Thanks to its great Java interoperability and to Kotlin extensions, Reactor Kotlin APIs leverage regular Java APIs and are additionally enhanced by a few Kotlin specific APIs available out of the box within Reactor artifacts.

For example, Kotlin reified type parameters provide a workaround for JVM generics type erasure, and Reactor provides some extensions to take advantage of this feature.

You can see bellow a quick comparison of Reactor with Java versus Reactor with Kotlin + extensions.

Reactor KDoc API lists and documents all the Kotlin extensions available.

One of Kotlin’s key features is null-safety - which cleanly deals with null values at compile time rather than bumping into the famous NullPointerException at runtime. This makes applications safer through nullability declarations and expressing "value or no value" semantics without paying the cost of wrappers like Optional. (Kotlin allows using functional constructs with nullable values; check out this comprehensive guide to Kotlin null-safety.)

Although Java does not allow one to express null-safety in its type-system, Reactor now provides null-safety of the whole Reactor API via tooling-friendly annotations declared in the reactor.util.annotation package. By default, types from Java APIs used in Kotlin are recognized as platform types for which null-checks are relaxed. Kotlin support for JSR 305 annotations + Reactor nullability annotations provide null-safety for the whole Reactor API to Kotlin developers, with the advantage of dealing with null related issues at compile time.

The JSR 305 checks can be configured by adding the -Xjsr305 compiler flag with the following options: -Xjsr305={strict|warn|ignore}.

For kotlin versions 1.1.50+, the default behavior is the same to -Xjsr305=warn. The strict value is required to have Reactor API full null-safety taken in account but should be considered experimental since Reactor API nullability declaration could evolve even between minor releases and more checks may be added in the future).

Suggest Edit to "Kotlin support"

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

This situation 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:

In order to test it, you want to verify the following scenario:

In the StepVerifier API, this translates to the following test:

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 their 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, often with verify() or one of its variants.

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.

StepVerifier can be used with time-based operators to avoid long run times for corresponding tests. This is done through the StepVerifier.withVirtualTime builder.

It looks like the following example:

This virtual time feature 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 prerequisite 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 guides you into 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.

verifyThenAssertThat() returns a StepVerifier.Assertions object, which you can use to assert a few elements of state once the whole scenario has played out successfully (because 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 information about the Context, see Adding a Context to a Reactive Sequence.

StepVerifier comes with a couple of expectations around the propagation of a Context:

Additionally, one can associate a test-specific initial Context to a StepVerifier by using StepVerifierOptions to create the verifier.

These features are demonstrated in the following snippet:

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 value or multiple values from the TestPublisher.Violation enum. The values define which parts of the specification the publisher can overlook. These enum values include:

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

When building complex chains of operators, you could come across cases where there are several possible execution paths, materialized by distinct sub-sequences.

Most of the time, these sub-sequences produce a specific-enough onNext signal that you can assert it was executed by looking at the end result.

For instance, consider the following method, which builds a chain of operators from a source and uses a switchIfEmpty to fallback to a particular alternative if the source is empty:

It is easy enough to test which logical branch of the switchIfEmpty was used, as follows:

But think about an example where the method produces a Mono<Void> instead. It waits for the source to complete, performs an additional task, and completes. If the source is empty, a fallback Runnable-like task must be performed instead, as follows:

In order to verify that your processOrFallback indeed goes through the doWhenEmpty path, you need to write a bit of boilerplate. Namely you need a Mono<Void> that:

Before version 3.1, you would need to manually maintain one AtomicBoolean per state you wanted to assert and attach a corresponding doOn* callback to the publisher you wanted to evaluate. This could be a lot of boilerplate when having to apply this pattern regularly. Fortunately, since 3.1.0 there’s an alternative with PublisherProbe, as follows:

You can also use the probe in place of a Flux<T> by calling .flux() instead of .mono(). For cases where you need to probe an execution path but also need the probe to emit data, you can wrap any Publisher<T> using PublisherProbe.of(Publisher).

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 are not very helpful. They take us on a travel inside the internals of what seems to be a reactive chain, through multiple calls to subscribe and request.

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 gets several rows in the trace, but overall these three classes are involved). So a range().flatMap().single() chain maybe?

But what if we use that pattern a lot in our application? This still does not 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, as follows:

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 remain unsure of which one caused the issue.

What we want to find out more easily is where the operator was added into the chain - that is, 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 Hooks.onOperator hook at application start (or at least before the incriminated Flux or Mono can be instantiated), as follows:

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. We call this captured assembly information a traceback.

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

When we reuse our initial example but activate the operatorStacktrace debug feature, the stack trace is as follows:

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 the section of the chain of operators that gets notified of the error:

We deal 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 or Mono upstream of it (including onNext, onError, and onComplete and 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 into how it works and what kind of events it propagates upstream to the range, as shown in the following example:

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 can 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: hot and cold.

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, do not 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 replays it to anybody subscribing to it later. 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).

Consider two other examples. The following code shows the first example:

This first example produces the following output:

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 the second example, shown in the following code:

The second example produces the following output:

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 and 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 versus subscriptions to the original source. These additional methods include the following:

Consider the following example:

The preceding code produces the following output:

With autoConnect:

The preceding code produces the following output:

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 a Collection<T>.

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

To obtain a ParallelFlux, you can use the parallel() operator on any Flux. By itself, this method does not parallelize the work. Rather, it divides 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 examples, the first of which is shown in the following code:

The following code shows the second example:

The first example produces the following output:

The second correctly parallelizes on two threads, as shown in the following output:

If, once you process 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>> through the groups() method and apply additional operators to them through the composeGroup() method.

As we have seen in the Threading and Schedulers section, Reactor Core comes with several Scheduler implementations. While you can always create new instances through the new* factory methods, each Scheduler flavor also has a default singleton instance that is accessible through the direct factory method (such as Schedulers.elastic() versus Schedulers.newElastic()).

These default instances are the ones used by operators that need a Scheduler to work when you do not explicitly specify one. For example, Flux#delayElements(Duration) uses the Schedulers.parallel() instance.

In some cases, however, you might need to change these default instances with something else in a cross-cutting way, without having to make sure every single operator you call has your specific Scheduler as a parameter. An example is measuring the time every single scheduled task takes by wrapping the real schedulers, for instrumentation purposes. In other words, you might want to change the default Schedulers.

Changing the default schedulers is possible through the Schedulers.Factory class. By default, a Factory creates all the standard Scheduler through similarly named methods. Each of these can be overridden with your custom implementation.

Additionally, the Factory exposes one additional customization method: decorateExecutorService. It is invoked during the creation of every reactor-core Scheduler that is backed by a ScheduledExecutorService (even non-default instances, such as those created by calls to Schedulers.newParallel()).

This lets you tune the ScheduledExecutorService to be used: The default one is exposed as a Supplier and, depending on the type of Scheduler being configured, you can choose to entirely bypass that supplier and return your own instance or you can get() the default instance and wrap it.

Finally, there is a last customizable hook in Schedulers: onHandleError. This hook is invoked whenever a Runnable task submitted to a Scheduler throws an Exception (note that if there is an UncaughtExceptionHandler set for the Thread that ran the task, both the handler and the hook will be invoked).

Reactor has another category of configurable callbacks that are invoked by Reactor operators in various situations. They are all set in the Hooks class, and fall into three categories:

One of the big technical challenges encountered when switching from an imperative programming perspective to a reactive programming mindset lies in how you deal with threading.

Contrary to what you might be used to, in reactive programming, a Thread can be used to process several asynchronous sequences that run roughly at the same time (actually, in non-blocking locksteps). The execution can also easily and often jump from one thread to another.

This arrangement is especially hard for developers that use features dependent on the threading model being more "stable", such as ThreadLocal. As it lets you associate data with a thread, it becomes tricky to use in a reactive context. As a result, libraries that rely on ThreadLocal at least introduce new challenges when used with Reactor. At worst, they work badly or even fail. Using the MDC of Logback to store and log correlation IDs is a prime example of such a situation.

The usual workaround for ThreadLocal usage is to move the contextual data, C, along your business data, T, in the sequence, by using Tuple2<T, C> for instance. This does not look good and leaks an orthogonal concern (the contextual data) into your method and Flux signatures.

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.

As an illustration of how it looks like, here is a very simple example of both writing to the Context and reading from it:

In the following sections, we’ll learn about the Context and how to use it, so that you will eventually understand the example above.

In very specific cases, your application may deal with types that necessitate some form of cleanup once they’re not in use anymore. This is an advanced scenario, for example when you have reference counted objects or when you’re dealing with off heap objects. Netty’s ByteBuf is a prime example of both.

In order to ensure proper cleanup of such objects, you need to accommodate for it on a Flux-by-Flux basis, as well as in several of the global hooks (see Using Global Hooks):

This is needed because each hook is made with a specific subset of cleanup in mind, and users might want ie. to implement specific error handling logic in addition to cleanup logic within onOperatorError.

Note that some operators are less adapted to dealing with objects that need cleanup. For example, bufferWhen can introduce overlapping buffers, and that means that the discard "local hook" above might see a first buffer as being discarded and cleanup an element in it that is in a second buffer which is still valid.

Although Java does not allow expressing null-safety with its type system, Reactor now provides annotations to declare nullability of APIs, similar to those provided by Spring Framework 5.

Reactor leverages these annotations, but they can also be used in any Reactor-based Java project to declare null-safe APIs. Nullability of types used inside method bodies is outside of the scope of this feature.

These annotations are meta-annotated with JSR 305 annotations (a dormant JSR that is supported by tools like IntelliJ IDEA) to provide useful warnings to Java developers related to null-safety in order to avoid NullPointerException at runtime. JSR 305 meta-annotations allows tooling vendors to provide null-safety support in a generic way, without having to hard-code support for Reactor annotations.

They are also used by Kotlin which natively supports null-safety. See this dedicated section for more details.

The following annotations are provided in the reactor.util.annotation package:

Suggest Edit to "Advanced Features and Concepts"

Note: all of these methods except Mono#toFuture will throw an UnsupportedOperatorException if called from within a Scheduler marked as "non-blocking only" (by default parallel() and single()).

Suggest Edit to "Which operator do I need?"

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 zip static method or zipWith/zipWhen operators, which by definition need an element from each source to produce their output.

Using data-suppressing operators on sources of zip is thus problematic. Examples of data-suppressing operators include then(), thenEmpty(Publisher<Void>), ignoreElements() and ignoreElement(), when(Publisher…​).

Similarly, operators that use a Function<T,?> to tune their behavior, like flatMap, do need at least one element to be emitted for the Function to have a chance to apply. Applying these on an empty (or <Void>) sequence will never produce an element.

You can use .defaultIfEmpty(T) and .switchIfEmpty(Publisher<T>) to respectively replace an empty sequence of T with a default value or a fallback Publisher<T>, which could help avoid some of these situations. Note that this doesn’t apply to Flux<Void>/Mono<Void> sources, as you can only switch to another Publisher<Void>, which is still guaranteed to be empty. Here’s an example of defaultIfEmpty:

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.

Since 3.2.0.RELEASE, Reactor comes with such a retry baked in: Flux.retryBackoff.

For the curious, here is how to implement an exponential backoff with retryWhen 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. Each subscription gets its own worker, so all elements pushed to the corresponding subscriber are published on the same Thread.

Single threaded schedulers may be used to ensure thread affinity for different stages in the chain or for different subscribers.

Most logging frameworks allow contextual logging, letting users store variables that are reflected in the logging pattern, generally by way of a Map called the MDC ("Mapped Diagnostic Context"). This is one of the most recurring use of ThreadLocal in Java, and as a consequence this pattern assumes that the code being logged is tied in a one-to-one relationship with a Thread.

That might have been a safe assumption to make before Java 8, but with the advent of functional programming elements in the Java language things have changed a bit…​

Let’s take the example of a API that was imperative and used the template method pattern, then switches to a more functional style. With the template method pattern, inheritance was at play. Now in its more functional approach, higher order functions are passed to define the "steps" of the algorithm. Things are now more declarative than imperative, and that frees the library to make decisions about where each step should run. For instance, knowing which steps of the underlying algorithm can be parallelized, the library can use an ExecutorService to execute some of the steps in parallel.

One concrete example of such a functional API is the Stream API introduced in Java 8 and its parallel() flavor. Logging with a MDC in a parallel Stream is not a free lunch: one need to ensure the MDC is captured and reapplied in each step.

The functional style enables such optimizations, because each step is thread-agnostic and referentially transparent, but it can break the MDC assumption of a single Thread. The most idiomatic way of ensuring any kind of contextual information is accessible to all stages would be to pass that context around through the composition chain. During the development of Reactor we encountered the same general class of problem, and we wanted to avoid this very hands-down and explicit approach. This is why the Context was introduced: it propagates through the execution chain as long as Flux and Mono are used as the return value, by letting stages (operators) peek at the Context of their downstream stage. So instead of using ThreadLocal, Reactor offers this map-like object that is tied to a Subscription and not a Thread.

Now that we’ve established that MDC "just working" is not the best assumption to make in a declarative API, how can we perform contextualized log statements in relation to events in a Reactive Stream (onNext, onError, and onComplete)?

This entry of the FAQ offers a possible intermediate solution when one wants to log in relation to these signals in a straightforward and explicit manner. Make sure to read the Adding a Context to a Reactive Sequence section beforehand, and especially how a write must happen towards the bottom of the operator chain for operators above it to see it.

To get contextual information from the Context to the MDC, the simplest way is to wrap logging statements in a doOnEach operator with a little bit of boilerplate code. This boilerplate depends on both the logging framework/abstraction of your choice and the information you want to put in the MDC, so it has to be in your codebase.

The following is an example of such a helper function around a single MDC variable and focused on logging onNext events, using Java 9 enhanced Optional API:

Using this boilerplate code ensures that we are good citizens with the MDC: we set a key right before we execute a logging statement and remove it immediately after. There is no risk of polluting the MDC for subsequent logging statements.

Of course, this is a suggestion. You might be interested in extracting multiple values from the Context or in logging things in case of onError. You might want to create additional helper methods for these cases or craft a single method that makes use of additional lambdas to cover more ground.

In any case, the usage of the preceding helper method could look like the following reactive web controller:

In this configuration, the restaurantService can emit its data on a shared thread, yet the logs will still reference the correct X-UserId for each request.

For completeness, we can also see what an error-logging helper could look like:

Nothing much has changed, except for the fact that we check that the Signal is effectively an onError, and that we provide said error (a Throwable) to the log statement lambda.

Applying this helper in the controller is very similar to what we’ve done before:

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

The reactor.function package contains functional interfaces that complement the Java 8 Function, Predicate and Consumer interfaces, for 3 to 8 values.

TupleUtils offers static methods that act as a bridge between lambdas of these functional interfaces to a similar interface on the corresponding Tuple.

This allows to easily work with independent parts of any Tuple, for instance:

Can be instead written as

The reactor.math package contains a MathFlux specialized version of Flux that offers mathematical operators, like max, min, sumInt, averageDouble…​

The reactor.retry package contains utilities to help in writing Flux#repeatWhen and Flux#retryWhen functions. The entry points are factory methods in respectively Repeat and Retry interfaces.

Both interfaces can be used as a mutative builder AND are implementing the correct Function signature to be used in their counterpart operators.

Since 3.2.0, one of the most advanced retry strategies offered by these utilities is also part of the reactor-core main artifact directly: exponential backoff is available as the Flux#retryBackoff operator.

Reactor-extra comes with several specialized schedulers:

Suggest Edit to "Reactor-Extra"