Repository files navigation

Non-BlockingReactive Streams Foundation for the JVM both implementing aReactive Extensions inspired API and efficient event streaming support.

Since3.3.x, this repository also containsreactor-tools, a java agent aimed at helping with debugging of Reactor code.

Reactor 3 requires Java 8 or + to run.

With Gradle from repo.spring.io or Maven Central repositories (stable releases only):

repositories {    mavenCentral()// Uncomment to get access to Snapshots// maven { url "https://repo.spring.io/snapshot" }}dependencies {    compile"io.projectreactor:reactor-core:3.8.0-M7"    testCompile"io.projectreactor:reactor-test:3.8.0-M7"// Alternatively, use the following for latest snapshot artifacts in this line// compile "io.projectreactor:reactor-core:3.8.0-SNAPSHOT"// testCompile "io.projectreactor:reactor-test:3.8.0-SNAPSHOT"// Optionally, use `reactor-tools` to help debugging reactor code// implementation "io.projectreactor:reactor-tools:3.8.0-M7"}

See thereference documentationfor more information on getting it (eg. using Maven, or on how to get milestones and snapshots).

Note about Android support: Reactor 3 doesn't officially support nor target Android.However it should work fine with Android SDK 21 (Android 5.0) and above. See thecomplete notein the reference guide.

Since the introduction ofJava Multi-Release JAR Filesupport one needs to have JDK 8, 9, and 21 available on the classpath. All the JDKs shouldbe automaticallydetectedorprovisionedby Gradle Toolchain.

However, if you see error message such asNo matching toolchains found for requested specification: {languageVersion=X, vendor=any, implementation=vendor-specific} (whereX can be 8, 9 or 21), it means that you need to install the missing JDK:

In the project root folder runSDKMAN env initialization:

sdk env install

then (if needed) install JDK 9:

sdk install java$(sdk list java| grep -Eo -m1'9\b\.[ea|0-9]{1,2}\.[0-9]{1,2}-open')

then (if needed) install JDK 21:

sdk install java$(sdk list java| grep -Eo -m1'21\b\.[ea|0-9]{1,2}\.[0-9]{1,2}-open')

When the installations succeed, try to refresh the project and see that it builds.

The manual Operation-system specific JDK installationis well explained in theofficial docs

The current active shell JDK version must be compatible with JDK17 or higher for Antora to build successfully.So, just ensure that you have installed JDK 21, as described above and make it as the current one.

Then you can build the antora documentation like this:

./gradlew docs

The documentation is generated indocs/build/site/index.html and indocs/build/distributions/reactor-core-<version>-docs.zipIf a PDF file should also be included in the generated docs zip file, then you need to specify-PforcePdf option:

./gradlew docs -PforcePdf

Notice that PDF generation requires theasciidoctor-pdf command to be available in the PATH.For example, on Mac OS, you can install such command like this:

brew install asciidoctor

New to Reactive Programming or bored of reading already ? Try theIntroduction to Reactor Core hands-on !

If you are familiar with RxJava or if you want to check more detailed introduction, be sure to checkhttps://www.infoq.com/articles/reactor-by-example !

A Reactive Streams Publisher with basic flow operators.

• Static factories on Flux allow for source generation from arbitrary callbacks types.
• Instance methods allows operational building, materialized on each subscription (Flux#subscribe(), ...) or multicasting operations (such asFlux#publish andFlux#publishNext).

Flux in action :

Flux.fromIterable(getSomeLongList())    .mergeWith(Flux.interval(100))    .doOnNext(serviceA::someObserver)    .map(d ->d *2)    .take(3)    .onErrorResume(errorHandler::fallback)    .doAfterTerminate(serviceM::incrementTerminate)    .subscribe(System.out::println);

A Reactive Streams Publisher constrained toZERO orONE element with appropriate operators.

• Static factories on Mono allow for deterministiczero or one sequence generation from arbitrary callbacks types.
• Instance methods allows operational building, materialized on eachMono#subscribe() orMono#get() eventually called.

Mono in action :

Mono.fromCallable(System::currentTimeMillis)    .flatMap(time ->Mono.first(serviceA.findRecent(time),serviceB.findRecent(time)))    .timeout(Duration.ofSeconds(3),errorHandler::fallback)    .doOnSuccess(r ->serviceM.incrementSuccess())    .subscribe(System.out::println);

Blocking Mono result :

Tuple2<Instant,Instant>nowAndLater =Mono.zip(Mono.just(Instant.now()),Mono.delay(Duration.ofSeconds(1)).then(Mono.fromCallable(Instant::now)))  .block();

Reactor uses aScheduler as acontract for arbitrary task execution. It provides some guarantees required by ReactiveStreams flows like FIFO execution.

You can use or create efficientschedulersto jump thread on the producing flows (subscribeOn) or receiving flows (publishOn):

Mono.fromCallable( () ->System.currentTimeMillis() ).repeat()    .publishOn(Schedulers.single())    .log("foo.bar")    .flatMap(time ->Mono.fromCallable(() -> {Thread.sleep(1000);returntime; })            .subscribeOn(Schedulers.parallel())    ,8)//maxConcurrency 8    .subscribe();

ParallelFlux can starve your CPU's from any sequence whose work can be subdivided in concurrenttasks. Turn back into aFlux withParallelFlux#sequential(), an unordered join oruse arbitrary merge strategies via 'groups()'.

Mono.fromCallable( () ->System.currentTimeMillis() ).repeat()    .parallel(8)//parallelism    .runOn(Schedulers.parallel())    .doOnNext(d ->System.out.println("I'm on thread "+Thread.currentThread()) )    .subscribe()

To bridge a Subscriber or Processor into an outside context that is taking care ofproducing non concurrently, useFlux#create,Mono#create.

Flux.create(sink -> {ActionListeneral =e -> {sink.next(textField.getText());         };// without cancellation support:button.addActionListener(al);// with cancellation support:sink.onCancel(() -> {button.removeListener(al);         });    },// Overflow (backpressure) handling, default is BUFFERFluxSink.OverflowStrategy.LATEST)    .timeout(Duration.ofSeconds(3))    .doOnComplete(() ->System.out.println("completed!"))    .subscribe(System.out::println)

Most of this cool stuff uses bounded ring buffer implementation under the hood to mitigate signal processing difference between producers and consumers. Now, the operators and processors or any standard reactive stream component working on the sequence will be instructed to flow in when these buffers have free room AND only then. This means that we make sure we both have a deterministic capacity model (bounded buffer) and we never block (request more data on write capacity). Yup, it's not rocket science after all, the boring part is already being worked by us in collaboration withReactive Streams Commons on going research effort.

"Operator Fusion" (flow optimizers), health state observers, helpers to build custom reactive components, bounded queue generator, converters from/to Java 9 Flow, Publisher and Java 8 CompletableFuture. The repository contains areactor-test project with test features like theStepVerifier.

https://projectreactor.io/docs/core/release/reference/docs/index.html

https://projectreactor.io/docs/core/release/api/

https://github.com/reactor/lite-rx-api-hands-on

https://www.infoq.com/articles/reactor-by-example

https://github.com/reactor/head-first-reactive-with-spring-and-reactor/

• Everything to jump outside the JVM with the non-blocking drivers fromReactor Netty.
• Reactor Addons provide for adapters and extra operators for Reactor 3.

Powered byReactive Streams Commons

Licensed underApache Software License 2.0

Sponsored byVMware