reactor / .github

Template repository for reactor projects, includes typical README + labels

Reactor Project

Starting from 3.0, Reactor is now organized into multiple projects:

• reactor-core
• reactor-netty
• addons like reactor-extra or reactor-pool
• other more community-driven integrations like reactor-kafka and reactor-rabbitmq

A set of compatible versions for all these projects is curated under a BOM ("Bill of Materials") hosted under this very repository.

Using the BOM with Maven

In Maven, you need to import the bom first:

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bom</artifactId>
            <version>2024.0.0-M1</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

Notice we use the <dependencyManagement> section and the import scope.

Next, add your dependencies to the relevant reactor projects as usual, except without a <version>:

<dependencies>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-core</artifactId>
    </dependency>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Using the BOM with Gradle

Gradle 5.0+

Use the platform keyword to import the Maven BOM within the dependencies block, then add dependencies to your project without a version number.

dependencies {
     // import BOM
     implementation platform('io.projectreactor:reactor-bom:2024.0.0-M1')

     // add dependencies without a version number
     implementation 'io.projectreactor:reactor-core'
}

Gradle 4.x and earlier

Gradle versions prior to 5.0 have no core support for Maven BOMs, but you can use Spring's gradle-dependency-management plugin.

First, apply the plugin from Gradle Plugin Portal (check and change the version if a new one has been released):

plugins {
    id "io.spring.dependency-management" version "1.0.11.RELEASE"
}

Then use it to import the BOM:

dependencyManagement {
     imports {
          mavenBom "io.projectreactor:reactor-bom:2024.0.0-M1"
     }
}

Then add a dependency to your project without a version number:

dependencies {
     compile 'io.projectreactor:reactor-core'
}

BOM Versioning Scheme

The BOM can be imported in Maven, which will provide a set of default artifact versions to use whenever the corresponding dependency is added to a pom without an explicitly provided version.

As the different artifacts versions are not necessarily aligned, the BOM represents a release train with an heterogeneous range of versions that are curated to work together. The artifact version follows the YYYY.MINOR.MICRO-QUALIFIER scheme since Europium, where:

• YYYY is the year of the first GA release in a given release cycle (like 3.4.0 for 3.4.x)
• .MINOR is a 0-based number incrementing with each new release cycle ** in the case of the BOM it allows discerning between release cycles in case two get first released the same year
• .PATCH is a 0-based number incrementing with each service release
• -QUALIFIER is a textual qualifier, which is omitted in the case of GA releases (see below)

On top of the artifact version, each release train has an associated codename, a chemical name from the Periodic Table of Elements in growing alphabetical order, for reference in discussions.

So far, the release trains code names are:

• Aluminium for the 3.0.x generation of Reactor-Core (💡)
• Bismuth for the 3.1.x generation (💡)
• Californium for the 3.2.x generation (💡)
• Dysprosium for the 3.3.x generation (💡)
• Europium (2020.0) for the 3.4.x generation (💡)

NOTE: Up until Dysprosium, the BOM was versioned using a release train scheme with a codename followed by a qualifier, and the qualifiers were slightly different. For example: Aluminium-RELEASE (first GA release, would now be something like YYYY.0.0), Bismuth-M1, Californium-SR1 (service release would now be something like YYYY.0.1), Dysprosium-RC1, Dysprosium-BUILD-SNAPSHOT (after each patch, we'd go back to the same snapshot version. would now be something like YYYY.0.X-SNAPSHOT so we get 1 snapshot per PATCH).

Contributing, Community / Support

As hinted above, this repository is for hosting the BOM and for transverse issues only. Most of the time, if you're looking to open an issue or a PR, it should be done in a more specific repository corresponding to one of the actual artifacts.

All projects follow the same detailed contributing guidelines which you can find here.

This document also give some ways you can get answers to your questions.

Documentation

• Guides
• Reactive Streams

Detail of Projects

Reactor Core

Reactive foundations for apps and frameworks and reactive extensions inspired API with Mono (1 element) and Flux (n elements) types

• API documentation: /docs/core/release/api

Reactor Netty

TCP and HTTP client and server.

• API documentation: /docs/netty/release/api

Reactor Addons

Extra projects adding features to reactor:

• reactor-adapter: adapt to/from various libraries, mainly RxJava 2.
  • API documentation: /docs/adapter/release/api
• reactor-extra: Retry utils, Math utils, ...
  • API documentation: /docs/extra/release/api

Snapshot Artifacts

While Stable Releases are synchronized with Maven Central, fresh snapshot and milestone artifacts are provided in the repo.spring.io repositories.

To add this repo to your Maven build, add it to the <repositories> section like the following:

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

To add it to your Gradle build, use the repositories configuration like this:

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

You should then be able to import a -SNAPSHOT version of the BOM, like 2020.0.{NUMBER}-SNAPSHOT for the snapshot of the {NUMBER}th service release of 2020.0 (Europium).

Sponsored by VMware

.github's Issues

Documents still refer to the Pivotal license header

bufferUntilChanged - emit last element of a hot publisher in its own buffer after a timeout

Motivation

Hi, I am encountering a somewhat weird behavior when using bufferUntilChanged with a hot publisher.
When an element comes in, it is compared to the previous element and if the keyComparator returns false, the previous element is added to the buffer and the buffer is emitted.
A new buffer is created containing the current element and it is left hanging until a new element arrives or the publisher completes.
If no new element arrives, this new buffer is not emitted.

More specifically, I am trying to implement a session window as defined by Kafka Streams and Flink, so I am actually comparing the event times of the elements.
This window is defined based on the time between two consecutive messages.
If the time between two consecutive messages is less than the specified session gap, then the messages are considered to belong to the same session.
If the gap is larger than the session gap, the window is emitted and a new window is started.

I created a test that shows this behavior, the last element is emitted in its own buffer only when the publisher is completed.

   void testBufferUntilChanged() {
        var testPublisher = TestPublisher.<Pair<Integer, Long>>create();
        var flux = testPublisher.flux().bufferUntilChanged(Function.identity(), (pair1, pair2) -> {
            var diff = pair2.getRight() - pair1.getRight();
            System.out.println("Diff " + diff);
            return diff < 500L;
        });
        
        StepVerifier
                .create(flux)
                .then(() -> testPublisher.next(Pair.of(1, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(400L))
                .then(() -> testPublisher.next(Pair.of(2, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(400L))
                .then(() -> testPublisher.next(Pair.of(3, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(600L))
                .then(() -> testPublisher.next(Pair.of(4, System.currentTimeMillis())))
                .assertNext(buffer -> assertThat(buffer.stream().map(Pair::getKey).collect(Collectors.toList()), containsInAnyOrder(1, 2, 3)))
                .thenAwait(Duration.ofMillis(1000L))
                .then(testPublisher::complete)
                .assertNext(buffer -> assertThat(buffer.stream().map(Pair::getKey).collect(Collectors.toList()), containsInAnyOrder(4)))
                .verifyComplete();
    } 

Desired solution

As an enhancement to the bufferUntilChanged methods, do you think it would be useful to create a new signature with a new Duration parameter?
It would be similar to other buffer methods that have Duration parameters.
If a new element does not arrive in the specified Duration interval, the buffer will be emitted no matter what.

Considered alternatives

As a workaround, I am thinking of adding some kind of scheduled heartbeat elements to the publisher.
I am using Flux.create, so I can emit items from a scheduled service.
But it is not ideal, I believe it will complicate the processing pipeline unnecessarily.
Thanks.

Additional context