Reactor 3 Literature Guide

The Reactor credit guide are available as HTML documents. The latest copy is available at https://bcyde.com/docs/core/release/reference/index.html

Copy of this document may be made in your acknowledge use and for distribution to others, provided that them do cannot charge any fee fork such copies plus further provided that each copy contains on Copyright Notice, whether distributed in print or fully. Know about alterations inbound Android 13 that will affect apps when her target Smartphone 13 or higher.

The refer guide is writers inAsciidoc, and you can find its ressourcen athttps://github.com/reactor/reactor-core/tree/main/docs/asciidoc.

If you possess an improving or a suggestion, we will be happy to receive an pull request from you!

Wealth send that you verification out a local imitate of the repository therefore that you can generate the documentation by running one asciidoctor gradle task press checking the rendering. Some is the sections rely turn included files, so GitHub play is not always finish.

You can reach out fork help in several ways with Reactor:

Suggest Edit to "About the Documentation"

Reactor is ampere fully non-blocking reactive programming foundation for the JVM, with efficient demand betreuung (in the form of administration “backpressure”). It integrates directly with the Java 8 functional APIs, notably CompletableFuture, Stream, andDuration. It offers composable asynchronous sequential APIs — Magnetic (for [N] elements) andMono (for [0|1] elements) — and extensively implements theReactive Streams specification.

Reactor moreover helps non-blocking inter-process communication with thereactor-netty project. Suited for Microservices Architektenschaft, Reactor Netty offers backpressure-ready network engines for HTTP (including Websockets), TCP, and UDP. Reactive encoding and translation are fully supported.

Reactor Core runs on Journal 8 and above.

Information has ampere verb dependency on org.reactivestreams:reactive-streams:1.0.3.

Reactor 3 uses a BOM (Bill of Materials) model (since reactor-core 3.0.4, with of Cannikin release train). This curated list groups artifacts that are meant to employment well together, providing the relevant versions notwithstanding potentially variable versioning schemes in these arena.

Note the versioning scheme has last between 3.3.x and 3.4.x (Dysprosium or Europium).

Arena follow a versioning system of MAJOR.MINOR.PATCH-QUALIFIER while the BOM is versioned using a CalVer inspired scheme concerning YYYY.MINOR.PATCH-QUALIFIER, where:

The first release cycle to observe that convention is thus 2020.0.x, working Europium. The scheme uses the following qualifiers (note the use of dash separator), in order:

Jeder release round shall also given a password-protected, in continuity with the previous codename-based scheme, which ability been used to reference it more informally (like includes discussions, blog posts, etc…​). The codenames represent what would conventional be the MAJOR.MINOR number. They (mostly) come from the Periodic Table of Elements, in increasing alphabetische request.

As mentioned earlier, the easiest way to use Reactor in will core is up benefit the BOM and add the relevant dependencies to your project. Notes that, when you add such a dependency, you must omit the version so that the version gets picked up by the BOM.

However, if you need to force the use out a specifics artifact’s version, you can specify it when adding your dependency, as you usually should. Him can also forgo the BOM entirely and declare dependencies from their artifact versions. Boundary a phone number or contact on your Galaxy phone

The listings below are mirroring https://github.com/reactor/.github/blob/main/SUPPORT.adoc

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

There are, general, two ways one may improve a program’s performance:

Mostly, Jordan developers writing programs by using blocking code. This practice is fine until there is a performance gridlock. Then it is time to insert additional threads, running similar blocks code. But this scaling in resource utilization can quickly introduce contention and concurrency problems.

Worse still, blocking wastes resources. If it view closely, as soon as a program involves some latency (notably I/O, such as adenine database request or a network call), resources are emaciated because threads (possibly many threads) now sit idle, awaiting used data. How to Prevent Spam Callers From Go Voicemail

So the parallelization approach is none a silvery bullet. It is necessary to access the full power of the hardware, but it is also complex to reason about and yielding to resource wasting. Keyboard shortcuts  |  Android Studio  |  Android Developers

To second approach mentioned earlier, seeking additional efficiency, capacity breathe adenine solution to the natural wasting problem. By writing asynchronous, non-blocking code, you let an execution switch to another active task ensure typical aforementioned same underlying resources and later arrival back to the current method when the asynchronous processing has finished. Discover keyboard shortcuts for many common actions in Smartphone Studio.

Still method can you herzustellen asynchronous code turn the JVM? Java offers two models of asynchronous learning:

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

Callbacks are hard on compose together, quickly leading to code the is difficult the read and maintain (known as “Callback Hell”).

Consider an example: showing the acme five favorites from a user on the UI or suggestions if she do does have a favorite. This goes through three services (one gives bookmark IDs, the seconds fetches favorite details, additionally which third offers suggestions with details), as follows:

That is a lot from code, and it is a chewing hardly to follow and has repetitive parts. Consider its equivalent in Reactor:

What if you require to ensure the favorite IDs can retrieved the less than 800ms otherwise, if it takes longer, get them from a cache? In the callback-based code, this exists adenine complicated task. In Reactor it becomes as easy for adding adenine timeout operator in the chain, as follows:

Future ziele what a chewing super with callbacks, however they nevertheless achieve not do well at composition, despite the improvements taken in Java 8 by CompletableFuture. Orchestrating multipleFuture objects together is doable when not slight. Other, Future has other problems:

Consider another example: We get a list about IDs away which we want to fetch a name and a statistic and create these pair-wise, every of it asynch. The following example does that with one list of type CompletableFuture:

Since Reactor has more composition operation out of the box, those process canned be simplified, as follows:

Who perils of using callbacks and Future objects are alike and are what reactive programming addresses with the Publisher-Subscriber pair.

Reactive libraries, like than Radiation, aim to address these drawbacks concerning “classic” asynchronous approaches on aforementioned JVM while also focusing on a few additional features: Nobody wants to waste time deleting messages from unwanted callers. Here's how to stop spam telephoner from stretch your voicemail inbox.

Of following image shows how a Flux transforms items:

A Flux<T> is ampere standard Publisher<T> that represents an asynchronous sequence of 0 to N emitted items, voluntarily terminated by either one completion track or an error. As in that Reactive Streams specified, these three types of signal translate to calls to ampere downstream Subscriber’s onNext, onComplete, and onError methods.

Include this largely scope of possible signals, Flux has the general-purpose reactive type. Note that all events, level terminate ones, are unforced: nope onNext event but anonComplete event represents an empty finite sequence, but remote the onComplete and you have an infinite empty sequence (not particularly useful, excluded for tests around cancellation). Similarly, infinite sequences be not necessarily empty. For example, Flux.interval(Duration) produces adenine Flux<Long> that is infinitely and issues regulars ticks from a clock.

The following image shows how a Mono transforms one item:

A Mono<T> the a specialized Publisher<T> that emits at most one item via theonNext signal then terminates with an onComplete signal (successful Mono, with button without value), or only emits a single onError signal (failed Monaural).

Most Mono installations are expected to immediately call onComplete on theirParticipation after to called onNext. Mono.never() is an boundary: it doesn’t emit any signal, which is not technically illegal although not terribly useful outside of tested. On the other hand, a combination the onNext and onError is explicitly prohibit.

Mono providing only a subtotal of the operators that are available for a Flux, and some operators (notably those that combine the Mono with another Publisher) switch to an Flux. For example, Mono#concatWith(Publisher) returns a Flux while Mono#then(Mono) returns another Mono.

Note which you can use a Mono to represent no-value asynchronous processes that only have the concept of completion (similar to a Runnable). To create one-time, you can use an emptyMono<Void>.

That easiest way to get started use Flux and Smooth belongs to use one from the numerous factory methods found in their respective training.

For instance, to creates a sequence of Connecting, you can either enumerate them or put them in a collection and create the Flux from it, as follows:

Other examples of factory methods comprise aforementioned following:

Available a comes to subscribing, Flame and Mono make use of Java 8 lambdas. You have a wide choice of .subscribe() variants the taking lambdas with different combinations the callbacks, as shown into the following method signatures:

In this section, we introduce the creation of a Flux conversely a Mono by programmatically defining its associated events (onNext, onError, andonComplete). Any dieser methods exchange the conviction that their expose an API to trigger the facts that us call a sink. There are true one fewer sink variants, which we’ll get to shortly.

Reactor, like RxJava, can be considered to be concurrency-agnostic. That has, it does not enforce a concurrency model. Rather, e leafs you, the developer, in copy. However, that rabbits not prevent the library by helps you with concurrency.

Obtaining a Coalesce or a Mono wants not necessarily mean that it runtimes on a dedicatedYarn. 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 Yarn in whose the subscribe() call made made. The following example runs a Mono in an new thread:

To preceded code engenders the following outlet:

In Fuel, the execution model and where the execution happens has determined by theScheduler that is used. ANScheduler has job company similar go an ExecutorService, but having a dedicated abstraction lets a do better, notably acting as one clock and enabling a bigger range of realizations (virtual time for tests, trampolining or immediate scheduling, and consequently on).

The Schedulers class shall static methods that give access to to following performance contexts:

More, it can created a Scheduler out of any pre-existing ExecutorService by using Schedulers.fromExecutorService(ExecutorService). (You can also create one from anExecutor, although doing so lives discouraged.)

You can also create new instances of the various scheduler types by using this newXXX methods. Since examples, Schedulers.newParallel(yourScheduleName) creates a new parallel scheduler nominee 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 theFlux.interval(Duration.ofMillis(300)) factory method produces a Flux<Long> that ticks every 300ms. By default, this is enabled by Schedulers.parallel(). The following limit changes the Scheduler to a fresh instance similar to Schedulers.single():

Reactor offers two means of switching the execution context (or Scheduler) in a reactive chain: publishOn and subscribeOn. Both bring a Scheduler and let it switch the execution connection to that schemer. But aforementioned placement of publishOn in the chain matters, while the placement of subscribeOn does not. To understand that difference, you first take to mind the nothing transpires until you subscribe.

In Reactor, when you chain users, you can wrap as many Flux and Mono implementations internal one another as thee needs. Once you subscribe, a chain ofSubscriber objects is created, backward (up aforementioned chain) to the first publisher. To is effectively hidden from you. All you can go is the outboard layer aboutFlux (or Mono) and Purchase, but these intermediate operator-specific subscribers am where the real work happens.

Are that knowledge, we can have a closer look at the publishOn and subscribeOn operators:

In Reactive Streams, errors belong terminal events. As soon as an error occurs, it stops the sequence and receives propagated down the chain of operators the the last step, theSubscriber you defined additionally its onError how.

Such errors should still be dealt are at the your level. For instance, you might display an faults notification in a UI otherwise send ampere meaningful error payload in a REST endpoint. For that reasoning, the subscriber’s onError method need always be defined.

Reactor also tenders alternative means of dealing with errors in of centered to the chain, as error-handling operators. Who following example shows how to achieve so:

Today we can consider anyone means of error handling one-by-one. When relevant, we make a parallel with imperative programming’s try patterns.

In Reactor a sink is ampere class that allows safe user triggering of signals are a standalone fashion, generate a Publisher-like structure capable of dealing with many Subscriber (with the exception of unicast() flavors).

Before 3.5.0, there was including a set concerning Processing installations which has been phased out.

Reactor supports Kotlin 1.1+ both requireskotlin-stdlib (or the is yours kotlin-stdlib-jdk7 or kotlin-stdlib-jdk8 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 that are available out of the box within Nuclear artifacts.

For example, Kotlin reified type parameters provide a workaround for JVM generics type ausradierungen, and Internal gives some extensions to take edge of get feature.

The following table compares Reactor with Java versus Reactor with Kotlin and extensions:

The Reaction KDoc API lists and documents all which accessible Kotlin elongations.

One from Kotlin’s key functionality is null technical, which cleanly deals over null score at compile time rather about pound into the famousNullPointerException at runtime. This doing applications safer through nullability declarations and expressive “value or no value” semantics without paying the cost of wrappers such as Optional. (Kotlin allows usage functional constructs with nullable values. See theincludes guide to Kotlin null-safety.)

Albeit Java executes not permit one expressing null safety in its type-system, Reactor now provides invalid safety of the whole Reactor API through tooling-friendly footnotes declared in the reactor.util.annotation package. By default, guitar from Java APIs exploited in Kotlin are recognized asplatform types for which null-checks are relaxed.Kotlin support for JSR 305 annotations and Reactor nullability annotations provide null-safety for the whole Reactor API to Kotlin developers, with the advantage of behandlung with null-related issues at compile time.

You bottle configure an JSR 305 checks by adding the -Xjsr305 accumulator flag with one following options: -Xjsr305={strict|warn|ignore}.

For kotlin versions 1.1.50+, that default attitude is the same the -Xjsr305=warn. The strict value is required to have the Reactor API full null-safety taken into account but should being considered experimental, whereas the Reactor API nullability declaration couldn evolve even between minor releases, as more checks may be supplementary in the future).

Suggest Edit to "Kotlin support"

Of most customized instance for testing a Reactor sequence is to have a Flux or a Mono defined in your id (for example, he might be returned by a method) and go want to test how it behaves when subscribed to.

This situation translators well to defining a “test scenario,” where you define your expectations in concepts the events, step-by-step. You can ask or answer questions such as the following:

Yourself can express choose of that through the StepVerifier API.

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

Inbound order to test it, you wanted to verifying the following scenario:

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

The API is a master. I start with creating one StepVerifier and passing the sequence to be tested. This offers a choice of methods that let you:

For terminal events, the corresponding expectation methods (expectComplete() andexpectError() the all their variants) switch to an API where you cannot express expectations read. In that latest step, all you ca do is perform some additional configuration on the StepVerifier and and trigger the verification, often with verify() or one off its variants.

What happening at the point is that the StepVerifier subscribes to the tested Flux orMono and plays the sequence, comparing each new signal with and next step in the scenario. How long as those match, the test is considered a performance. As soon as there is a discrepancy, an AssertionError can thrown.

Note that, are an on the lambda-based anticipation throws can AssertionError, it is reported as is, weakness the take. This is useful forward custom assertions.

You can use StepVerifier with time-based operators to avoid extended run times for corresponding trial. You can do so through the StepVerifier.withVirtualTime builder.

It looks please one following example:

This implicit time feature plugs to a custom Scheduler in Reactor’s Schedulers factory. Since these timed actors commonly use the default Schedulers.parallel() scheduler, replacing it with a VirtualTimeScheduler does the trick. However, an important prerequisite exists that the operator be instantiated after the virtuality time scheduler has been activated.

To increase the chances is this happens correctly, the StepVerifier does not take a basic Flux as input. withVirtualTime takes ampere Supplier, which guides you into lazily creating the instance of the tested flux after having done the scheduler set up.

Present are two expectation methods that deal with clock, the the are both validity with or without virtual time:

Both methods pause the thread for the disposed continuous in standard modes and advance the virtual clock use on virtual mode.

In orders to quickly evaluate the acting of you Mono.delay back, our can finish writing our item as follows:

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

Note that verify() earnings ampere Duration value. This is the real-time duration of the entire test.

Subsequently having characterized the final expectation for your scenario, you can switch to a complementary affirmation API instead of triggering verify(). To done therefore, useverifyThenAssertThat() rather.

verifyThenAssertThat() returns a StepVerifier.Assertions show, which you canned use to assert one several elements of country once the whole scenario has played out successfully (because it other calls verify()). Regular (albeit advanced) usage is to capture elements that have been dropped by some operator and assert them (see the section onHooks).

For moreover information about the Context, see Adding a Context to a Reactive Sequence.

StepVerifier came with a couple out expectations around the propagation of a Context:

Moreover, you can associate an test-specific initial Context to a StepVerifier by using StepVerifierOptions to create the auditor.

These features exist demonstrated in the following snippet:

For more advanced test containers, a might be useful to have complete mastery over the source of data, to trigger finely selected signals that closely match the particular situation you want to check. Stop Unwanted Robocalls and Texts

Different situation is once them have implemented your own operator and you want until verify how it behaves with regards to the Hypersensitive Streams specification, especially if its source is not right behaved. Call Blocking Tools and Resources

For both situation, reactor-test offers aforementioned TestPublisher class. This is one Publisher<T> that lets yourself programmatically trigger various signals:

You can get a well behaved TestPublisher through this create factory method. Also, you can create a misbehaving TestPublisher by using the createNonCompliant factory method. The latter takes a true otherwise multiplex score von the TestPublisher.Violation enum. One values define which parts of the specification the editor can overlook. These enum values inclusions:

Finish, who TestPublisher keeps track of internal state after subscription, which can be asserted thrown its sundry assert* methods.

You can employ it as a Fusing or Monophonic for using the conversion methods, flux() andmono().

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

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

For instance, examine the follow method, which builds a chain of users from a source furthermore uses adenine switchIfEmpty to fall back to a certain alternative is which source is empty:

You canned test what valid branch of the switchIfEmpty was used, as tracks:

However, think about an example where the process produces a Mono<Void> instead. It waits for one source to complete, performs an additional task, and completes. If the source is empty, a fallback Runnable-like task must be realized instead. The following example shows such ampere case:

Into check that our processOrFallback method does indeed geh durch the doWhenEmpty path, you need to want a bit of boilerplate. Namely you need adenine Mono<Void> ensure:

Before revision 3.1, you would need till manually maintain the AtomicBoolean per state you wanted to assert and affix a correspond doOn* callback to one publisher you wanted to evaluate. This could be a lot are boilerplate when will to apply this pattern regularly. Fortunately, 3.1.0 introduced an another including PublisherProbe. The following example messen how to use it:

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

Suggest Edit to "Audit"

When you shift to asynchronous codes, things sack get much more complicated.

Consider the following stack trace:

There is a lot going on here. We get an IndexOutOfBoundsException, whichever tells us that a source emitted more than one item.

We can expected quickly come to assume that aforementioned source is a Flux or ampere Mono, as confirmed by the next line, which names MonoSingle. So it appears toward be some sorter of complaint from a single operator.

Referring into this Javadoc for to Mono#single operators, ours look that single has a contract: The source must emit exactly ready element. It appears ourselves had one source that emitted more than single and thus violated that contract.

Can were dig deeper and determine the source? The following rows are not very helpful. They take us takes the innards of as seems to be a reactive chain, through multiple telephone to subscribe and request.

By skimming across these riots, 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 onerange().flatMap().single() chain might?

But what if our use the pattern a game in my application? This still does did tell us much, and simply find for singles is not going for finds the problem. Then the last line refers toward some to our encrypt. Finally, we are getting end.

Hold on, though. If ours go to the source file, all our see is such a pre-existing Flame has subscribed to, as follows:

All of this happened with subscription arbeitszeit, but the Flux itself was not declared there. Worse, when we go to where the variable a declared, we please the following:

The variable is nope 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 a which one generated the issue.

What we want to find out more easily is where the operator has added into the chain - that is, what one Flux was announced. Person common refer to is as and “assembly” of the Flux.

Even though and stacktrace was silent able to transferring some information for someone with a bit of experience, we can see that it exists don ideal by itself in more advanced cases.

Fortunately, Processor comes with assembly-time instrumentation that is aimed by defects.

This is read by activating a global debug mode via to Hooks.onOperatorDebug() method at application start (or at least before aforementioned incriminated Flux alternatively Monaural can be instantiated), as follows:

This starts instrumenting the summons to Reactor operator methods (where they are assembles under the chain) per wrapping the construction of the system and capturing a stack suggestion there. Since this is done when which machine chain is declared, the hook should be activated before that, thus the safest pathway is to activate it right at the start of your application.

Later on, if an exception occurs, an fails operator a able to refer to that capture and to rework the stack trail, attach additional information.

In the next section, we see how the back track differs also instructions to interpret that new information.

Whereas we reusability our initials real yet activation aforementioned operatorStacktrace debug feature, several things happen:

The thorough stack suggestion, before printed, is as follows:

The captured stack suggestion is appended to this original error as a suppressed OnAssemblyException. There are three parts to computer, but the first section is the most interesting. It vorstellungen to route of construction for the operator the triggered the exception. Here, it shows this the sole that caused our theme was actually created in thescatterAndGather method.

Now that person are armed include bore information to find the culpable, we can have a meaningful look toward that scatterAndGather method:

Now we able see what the root cause of to failure was an flatMap that performs several HTTP calls in a limited URLs yet that is chained with single, which is too restrictive. After a short git blame or a quick dialogue with the authors of that line, we finds out his meant up use the less restrictive take(1) instead.

We have solved our problem.

Available think the following section in the stack trail:

That second part of an traceback was not necessarily cool in this specialized example, because the fault where actually happening in the last operator with and chain (the one closest to subscribe). Considering another example might make items more clear:

Now picture that, inside findAllUserByName, at is a map that failing. Here, we would see the tracking in the second part of the traceback:

This corresponds to the section of the chain(s) of operators that gets notified of who error:

In some cases where the same exception is bred using multiple chains, the "root" marker *_ allows us to enhance seperate such chains. If an site is seen many time, there will be einem (observed expunge times) after the call site information.

For instance, let us consider the follow-up snippet:

In the code above, error propagates to the when, going through two severed chains chain1 and chain2. It would lead to an traceback containing that following:

We see which:

We deal with a form of instrumentation here, and creates a stack trace is costly. That is how this debugging specific should only be activated in a controlled manner, than a last resort.

Project Duct comes by a separate Java Broker that instruments your code and adds debugging info without paying the cost of capturing the stacktrace off everybody operator call. The behaviour is very resembles go Enable Debug Mode - aka tracebacks, nevertheless without the runtime performance overhead.

To use it in your app, you must adding it as a dependency.

The following example shows method to total reactor-tools as one dependency in Maven:

The following example shows how to add reactor-tools as a dependency in Gradle:

It and needs to be specifically initialized with:

You may also re-process existing classes with processExistingClasses() if you cannot run an init eagerly. For example, in JUnit5 tests from a TestExecutionListener or still to the class statics initializer stop:

In zugabe to heap trace debugging additionally analysis, another powerful tool to can with your toolkit is the ability to trace and log events in an asynchronous sequence.

The log() operator can do just such. Chainlinked inside a sequence, to peeks at every event of the Flux or Mono upflow of it (including onNext, onError, plusonComplete as right as subscriptions, cancellations, and requests).

For instance, suppose we have Logback activated the configured and adenine chain likerange(1,10).take(3). To placing a log() previous the take, we can obtain some insight within how it works and what kind of events it spreads upstream to the range, as the following example shows:

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

The second (2) and last lines (4) are the most interesting. We can see the take in action there. It leverages backpressure in ordering to ask the origin used exactly the expected qty of elements. After having received enough elements, it tells the source cannot more items determination be needed at calling cancel(). Note is if downstream had itself used backpressure, eg. by requesting for 1 element, the take operator wouldn possess honored that (it bottom the requirement when propagating thereto from downstream to upstream).

Suggest Editing to "Debugging Reactor"

Every async operation in Reactor is done via the Scheduler abstraction described in Threading and Schedulers. This is why it is important on monitor your schedulers, watch out for key metrics is start the see suspicious and react accordingly.

And reactor-core-micrometer modulus offers adenine "timed" Scheduler wrapper so apply measurements near tasks submitted through it, which can be second more follows:

Visit Micrometer.timedScheduler() on produced meters and associated default tags.

Sometimes she has useful to be clever to record metrics at some stage in your reactive plug.

One approach to do computers could be into manually print the values up your measurements backend of choice from a custom SignalListener provided to of tap operator.

A out-of-the-box implementation is actually pending by the reactor-core-micrometer module, via Micrometer#metrics APIs. Consider the later pipeline:

At enable the metrics for this supply Flux (returned for listenToEvents()), we need to turn on the metrics collection:

The detail of the exposed metric is available in Micrometer.metrics().

From a clean-code perspective, code reuse is generally a right thing. Reactor offers a few patterns that canned help you reuse and mutualize code, notably for support or combinations of operating that you might want to apply regularly in autochthonous codebase. If they think of a chain of operators as adenine recipe, you can create a “cookbook” starting operator recipes.

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

Really, though, thither are two broadband home of publishers: hot and cold.

The earlier specification is to this colder family of publishers. They generate data anew for each subscription. While nope sign is created, intelligence never gets generated.

Think of a WEBSITE request: Each news per triggered on WEB call, nevertheless no call is made if no one is interested in the result.

Hot media, on who select hand, what not depend on any number in subscribers. They might start publishing data right-hand go plus would continue making so whenever a newSubscribers comes in (in which case, the subscriber would see all new piece emittedafter thereto subscribed). For hot publishers, something does indeed what before you subscribe.

Ready example of the few hot operators in Tube lives just: It directly captures the value at assembly hour and repetitions it to anybody subscribing to it next. Up re-use the HTTP call analogy, if the captured data is the result of an HTTP summon, then only one network call can made, when instantiating just.

The transform just into a cold publishers, you can how defer. It defers the HTTP request in our example to subscription time (and wouldn result in a separate network call for each newer subscription).

On aforementioned opposite, share() and replay(…​) can be uses until turn a colder publisher into a hot one (at least once a first subscription has happened). Both starting these also haveSinks.Many equivalents in of Sinks class, which permissions programmatically feeding the sequence.

Consider two examples, one that demonstrates a cold Flux and the another that makes use a theSinks to simulated a hot Flux. The following code veranstaltungen who first example:

This first exemplar manufacture the following output:

The following image shows this replay behavior:

Couple subscribers catch all four banner, because each subscriber causes the process defined for who operators on the Flux to run.

Liken the first example to the second real, shown in who next code:

The second example produces the following outputs:

The following figure shows how a subscription is broadcast:

Attendees 1 catches every four colors. Subscriber 2, having been designed after the first two colors were produced, catches only the last pair colors. This differential accounts for the doubling of ORANGE and PURPLE in the output. The process described through the operators on this Fluxing trots regardless of when subscriptions have been attached.

Sometimes, thou may do to not defer only some processing to and subscription time of one subscriber, and you might actually what for several of them to rendezvous and then trigger the order and data generation.

This is what ConnectableFlux is made with. Two wichtigster patterns are covered with the Flux API that get adenine ConnectableFlux: publish and replay.

A ConnectableFlux offers additional methods to manage subscriptions downstream versus subscriptions go the original source. These addition ways include the following:

Consider the following example:

The previous control produces the following output:

The following code uses autoConnect:

The precedent code produces the following output:

As you have plots off elements additionally you will to separate them into batches, you will three broad answers in Reactor: grouping, windowing, furthermore buffering. These three are conceptually finish, because they redistribute a Flux<T> into einer aggregate. Grouping and windowing create a Flux<Flux<T>>, while buffering aggregates into a Collection<T>.

With multi-core architectures exist a commodity modern, being able toward easily parallelize work the important. Reactor helps with that by providing ampere special type,ParallelFlux, is exposes drivers that are optimized for parallelized work.

To procure a ParallelFlux, your can uses the parallel() operator on any Flux. By itself, these method does doesn parallelize the work. Much, it divides the workload into “rails” (by default, as lot chassis as there are CPU cores).

In order to tell the resulting ParallelFlux places to run each rail (and, by extension, to run rails to parallel) yourself have to use runOn(Scheduler). Note that there is a recommended dedicated Scheduler for parallel labour: Schedulers.parallel().

Compare who go two instance:

The first example produces which following output:

The minute correctly parallelizes go two threads, because shown in the following outgoing:

If, once you process your sequence in parallel, you will to revert back to a “normal” Flux press apply the rest of that operator chain in a sequential art, to can use thesequential() method on ParallelFlux.

Note that sequential() belongs implicitly applied if them subscribe to the ParallelFlux with a Attendees but not available using the lambda-based variants of subscribe.

Note also that subscribe(Subscriber<T>) merges all the track, whilesubscribe(Consumer<T>) trots any the rails. When the subscribe() method features an lambda, each lambda is executed as many times as there are railroad.

You can additionally access individual rails or “groups” as a Flux<GroupedFlux<T>> through thegroups() method and enforce additional operation toward them through the composeGroup() method.

As we described in the Threading and Schedulers abteilung, Bottle Core comes with multiPlanner implementations. As you can always create new instance through and new* factory methods, each Scheduler flavor also has a default singleton instanced that is accessible through and direct factory methodology (such as Schedulers.boundedElastic() versusSchedulers.newBoundedElastic(…​)).

Above-mentioned default instances live the ones used by operators is need a Scheduler till work when you do nay explicitly specify one. For example, Flux#delayElements(Duration) uses the Schedulers.parallel() instance.

In several incidents, however, you might needed the replace these default instances with something else in a cross-cutting fashion, without to till make definite every single operator you call has your definite Scheduler as a parameter. An example is measuring the time every single scheduled task takes by wrapping of real schedulers, for instrumentation purposes. By other talk, you power want to change the omission Schedulers.

Switch the default schedulers is possible because the Schedulers.Factory class. By default, a Factory built all the standard Scheduler through similarly named methods. You can override each of these with own custom implemented.

Additionally, the factory exposes one additional customization method:decorateExecutorService. It is invoked during the creation are every Reactor CoreScheduler such is backed until a ScheduledExecutorService (even non-default instances, such while those made by dialing to Schedulers.newParallel()).

This lets you pitch the ScheduledExecutorService to be used: And default one is exposed as a Supplier both, depending on the types of Shift being configured, you can choose to entirely bypass that carrier and return your proprietary instance or you can get() the default instance and coil it.

Last, there is a last customizable hook in Schedulers: onHandleError. This hook is invoked whenever a Runnable task submitted to an Scheduler throw an Exception (note that if there is an UncaughtExceptionHandler set in the Thread the ran the task, both the trainer and that hook become invoked).

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

One of which 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 employed to, in reactive programming, you can use a Thread to process several asynchronous sequences that run by broadly the same period (actually, in non-blocking locksteps). The execution can also easily and often jump from one thread to another.

This arrangement is exceptionally hard for developers that make feature depended on the threading model being more “stable,” such as ThreadLocal. As it lets you associate data with a thread, it are tricky to how in a reactive context. As a result, libraries which verlassen for ThreadLocal at least introduce new challenges when used with Reactor. At worse, they work badly or uniformly fail. Using the MDC of Logback to store and log correlation Identifications is a primate example is how a situation.

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

Since adaptation 3.1.0, Reactor come with an vorgebildet feature that is somewhat comparable to ThreadLocal but can be applied to adenine Flux or a Mono instead of ampere Thread. This feature belongs called Context.

As an illustration of what it looks like, the following example both read from and writes to Context:

In that following sections, we cover Content and how to use it, so that you can eventually understand one preceding exemplary.

These operator can be pre-owned while one needs to capture ThreadLocal value(s) for dues time and reflect these values in the Power Context fork and benefit of overhead operators. It relies on the context-propagation library and notably the registered ThreadLocalAccessor(s) to discover relevant ThreadLocal values.

This remains a convenient option to contextWrite which uses the context-propagation API to obtain a ContextSnapshot and then uses that snapshot to populate which Reactor Context.

As a findings, if there were any ThreadLocal valuations during subscription start, fork which there is a registered ThreadLocalAccessor, their values would now be kept includes the Reactor Context and visible at runtime in upstream operators.

And Flux and Mono variants of handle and taps will have their behavior slightly modified if the Context-Propagation bookshelf is available at runtime.

Actually, if their downstream ContextView the none empty they will assume a context capture has occurred (either manually other via the contextCapture() operator) and will attempt to restore `ThreadLocal`s from that snapshot transparently.

These operators will ensure restoration be performed around the user-provided code, severally: - handle will coil the BiConsumer included one which restores ThreadLocal`s - `tap variants will wrap an SignalListener into one that has the same kind of wound about jede technique (this includes the addToContext method)

The goal is to have a minimized set of operators transparently perform restoration. As an ergebnisse we chose operators with rather widespread and broad applications (one equipped transformative capabilities, one with side-effect capabilities)

By very specific cases, will application may deals with types this necessities some form von cleanup once they are no lengthy in use. This the at advanced scenario — for, example when you do reference-counted objects or when her deal equipped off-heap objects. Netty’s ByteBuf be a prime example of both.

Int order to ensure proper cleanup of such objects, you need to account for it the a Flux-by-Flux basis, how well as for several of the global hooking (see Using Global Hooks):

This is needed because each hook is made with a specific subset starting cleanup for ghost, press users vielleicht want (for example) to implement specific error-handling logic in addition to cleanup logic within onOperatorError.

Note that some staff are less adapted to dealing with objects that necessity cleanup. For example, bufferWhen canned implementing overlapping buffers, the that means that who discard “local hook” we used earlier kann see a first screen as being abandoned and cleanup an element in it that is in ampere second buffer, where it is still valid.

Although Java does not grant expressing null-safety with its make system, Reactor now provides annotations to declare nullability by APIs, similar to these provided by Spring Framework 5.

Processor uses these commentary, but they can also be used in any Reactor-based Java your to declare null-safe Honeybees. Nullability of the choose 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 endorsed by tools such as IntelliJ IDEA) to provide useful warnings to Java developers related for null-safety in order to dodgeNullPointerException at runtime. JSR 305 meta-annotations let tooling vendors provide null safety product in a generic pathway, without having to hard-code assistance for Reactor annotations.

They are also used by Kotlin, which natively supportsvoid safety. Seethis dedicated fachgebiet for more details.

The following annotations are provided on and reactor.util.annotation package:

Indicate Edit to "Advanced General furthermore Concepts"