Reactor Core

Reactor is a non-blocking reactive programming foundation for the JVM designed for efficient demand management (backpressure). It integrates with Java 8 functional APIs like CompletableFuture, Stream, and Duration.

Key components include:

• Flux: An API for asynchronous sequences of 0 to N elements.
• Mono: An API for asynchronous sequences of 0 or 1 element.
• Reactive Streams implementation: Extensively implements the Reactive Streams specification.
• reactor-netty: Supports non-blocking inter-process communication (HTTP, Websockets, TCP, UDP) with backpressure support.

Reactor Core provides two primary composable reactive types that implement the Publisher interface. These types are used to represent asynchronous sequences with different cardinalities:

• Flux: Represents a reactive sequence of 0 to N items.
• Mono: Represents a single-value-or-empty (0 to 1) result.

Choosing between them provides semantic clarity. For example, an HTTP request should be expressed as a Mono<HttpResponse> because it produces at most one response. Operators that change the cardinality of a stream will switch the type (e.g., calling .count() on a Flux returns a Mono<Long>).

A Mono<T> is a specialized Publisher<T> designed to represent an asynchronous 0-1 result. It follows these termination rules:

• Successful Mono: Emits at most one item via onNext and then terminates with an onComplete signal (the onNext signal is optional).
• Failed Mono: Emits a single onError signal.

Constraints:

• A combination of onNext and onError is explicitly forbidden.
• Mono.never() is a special case that emits no signal at all.
• To represent an asynchronous process that only has a concept of completion without a value (similar to a Runnable), use an empty Mono<Void>.

Backpressure is the mechanism that allows a Subscriber to signal to a Publisher that the rate of data emission is too high.

Reactor implements a push-pull hybrid model:

• The subscriber can request a specific number of elements (n) using the request(n) mechanism.
• Intermediate operators can modify these requests (e.g., a buffer operator might request 10 elements when the subscriber only requests 1).
• This prevents fast producers from overwhelming slow consumers.

As of Reactor Core 3.8.0, Project Reactor uses JSpecify annotations to declare the nullability of APIs, fields, and type usages. This allows for build-time nullability checks (e.g., using NullAway) and provides better integration with Kotlin's null safety system.

In Kotlin, JSpecify annotations are automatically translated to Kotlin's native null safety. In Java, they are used by IDEs (IntelliJ IDEA, Eclipse) and static analysis tools to prevent NullPointerException at runtime.

Reactor publishers fall into two categories:

• Cold Publishers: Generate data anew for each subscription. No data is generated until a subscriber is present. An example is an HTTP request where each subscriber triggers a new call.
• Hot Publishers: Do not depend on the number of subscribers. They may emit data regardless of whether anyone is listening. A new subscriber to a hot publisher will only see elements emitted after the subscription occurs.

Key transformations:

• Use Flux.defer() to turn a hot-like source (like Flux.just()) into a cold publisher by deferring execution until subscription time.
• Use .share() or .replay(...) to turn a cold publisher into a hot one (after the first subscription).

A Flux<T> is a Publisher<T> representing an asynchronous sequence of 0 to N items. It follows the Reactive Streams specification by emitting three types of signals to a downstream Subscriber:

1. onNext: Emits an item.
2. onComplete: Signals a successful termination of the sequence.
3. onError: Signals a failure and terminates the sequence.

Key characteristics:

• Empty Finite Sequence: A sequence with no onNext events but an onComplete event.
• Infinite Empty Sequence: A sequence with no onNext events and no onComplete event (typically used for testing cancellation).
• Infinite Non-empty Sequence: A sequence that emits items continuously without an onComplete signal (e.g., Flux.interval(Duration)).

Reactor implements the Reactive Programming paradigm, which focuses on asynchronous data streams and the propagation of change.

Key concepts include:

• Publisher-Subscriber Model: Unlike the pull-based Iterator pattern, reactive streams are push-based. A Publisher notifies a Subscriber of new values via onNext, or signals termination via onError (error) or onComplete (completion).
• Declarative Logic: You express what should happen to the data using operators, rather than describing the exact imperative control flow.
• Asynchronicity: Reactor allows for non-blocking code, enabling execution to switch to other tasks while waiting for I/O (like database or network calls), which improves resource efficiency compared to traditional blocking threads.

To use highlight.js on a web page, include the library, a CSS theme, and call hljs.initHighlightingOnLoad(). The library will automatically find and highlight code within <pre><code> tags by attempting to auto-detect the language.

To explicitly specify a language when auto-detection fails, use the class attribute on the <code> tag with the language name (e.g., class="html") or use the language- or lang- prefixes.

To disable highlighting for a specific block, use the nohighlight class.

<link rel="stylesheet" href="/path/to/styles/default.css">
<script src="/path/to/highlight.pack.js"></script>
<script>hljs.initHighlightingOnLoad();</script>

<!-- Explicit language specification -->
<pre><code class="html">...</code></pre>

<!-- Disabling highlighting -->
<pre><code class="nohighlight">...</code></pre>

For resilient retries that avoid overloading unstable systems, use Retry.backoff(maxAttempts, duration) within the retryWhen operator. This implements an exponential delay between attempts.

You can customize the behavior using:

• .jitter(double): Adds randomness to the delay.
• .doAfterRetry(Consumer<Retry): Executes an action after each retry attempt.
• .onRetryExhaustedThrow((spec, rs) -> Throwable): Defines which exception to throw when retries are exhausted.

Flux.<String>error(new IllegalStateException("boom"))
    .doOnError(e -> System.out.println(e))
    .retryWhen(Retry
            .backoff(3, Duration.ofMillis(100)).jitter(0d)
            .doAfterRetry(rs -> System.out.println("retried attempt " + rs.totalRetries()))
            .onRetryExhaustedThrow((spec, rs) -> rs.failure())
    );

Milestones and Release Candidates are available in Maven Central. However, Snapshots are distributed via the Spring Snapshots repository. You must add this repository to your build configuration to use snapshot versions.

<!-- Maven -->
<repositories>
	<repository>
		<id>spring-snapshots</id>
		<name>Spring Snapshot Repository</name>
		<url>https://repo.spring.io/snapshot</url>
	</repository>
</repositories>

// Gradle
repositories {
  maven { url 'https://repo.spring.io/snapshot' }
  mavenCentral()
}