Reactor Core

Non-Blocking Foundation for the JVM both implementing a inspired API and efficient event streaming support.

Getting it

Reactor 3 requires Java 8 or + to run.

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

repositories {//      maven { url 'http://repo.spring.io/snapshot' }      maven { url 'http://repo.spring.io/milestone' }      mavenCentral()    }    dependencies {      //compile "io.projectreactor:reactor-core:3.1.4.RELEASE"      //testCompile("io.projectreactor:reactor-test:3.1.4.RELEASE")      compile "io.projectreactor:reactor-core:3.2.0.M1"      testCompile("io.projectreactor:reactor-test:3.2.0.M1")    }

See the

for 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 26 (Android O) and above. See the

in the reference guide.

Getting Started

New to Reactive Programming or bored of reading already ? Try the !

If you are familiar with RxJava or if you want to check more detailled introduction, be sure to check

!

Flux

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 Flux#subscribe(), Flux#subscribe() or multicasting operations such as Flux#publish and Flux#publishNext.

Flux in action :

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

Mono

A Reactive Streams Publisher constrained to ZERO or ONE element with appropriate operators.

• Static factories on Mono allow for deterministic zero or one sequence generation from arbitrary callbacks types.
• Instance methods allows operational building, materialized on each Mono#subscribe() or Mono#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
     
       nowAndLater =         Mono.zip(                Mono.just(System.currentTimeMillis()),                Flux.just(1).delay(1).map(i -> System.currentTimeMillis()))            .block();
     

Schedulers

Reactor uses a as a

contract for arbitrary task execution. It provides some guarantees required by Reactive Streams flows like FIFO execution.

You can use or create efficient

to 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); return time; })            .subscribeOn(Schedulers.parallel())    , 8) //maxConcurrency 8    .subscribe();

ParallelFlux

can starve your CPU's from any sequence whose work can be subdivided in concurrent

tasks. Turn back into a Flux with ParallelFlux#sequential(), an unordered join or use abitrary 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()

Custom sources : Flux.create and FluxSink, Mono.create and MonoSink

To bridge a Subscriber or Processor into an outside context that is taking care of

producing non concurrently, use Flux#create, Mono#create.

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

The Backpressure Thing

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 with on going research effort.

What's more in it ?

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

---

Reference Guide

Javadoc

Getting started with Flux and Mono

Reactor By Example

Head-First Spring & Reactor

Beyond Reactor Core

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

---

Powered by

Licensed under

Sponsored by