reactor/reactor-core v3.1.0.RELEASE

on GitHub

This is the first release of Reactor 3.1, part of Bismuth-RELEASE Release Train.

This version is a recommended update for all Reactor 3 users, and will be the one backing Spring Framework 5.0. A lot of API polishing has occurred between the last 3.0.x release and this one, so please read the release notes below carefully if you didn't progressively upgrade through the various MILESTONEs and RCs. (if you only care about what changed since RC1, please have a look at the RC2 release note)

The release notes below are written from the perspective of a migration from 3.0.x latest version and cover
changes across all the 3.1.0 pre-releases.

Note that since pre-release v3.1.0.M2, the source code for the reactor-test artifact is part of the reactor-core github repository.

⚠️ Update considerations and deprecations

• Reactor has been updated to pass the TCK of the new Reactive Streams specification 1.0.1.RC2. A new StrictSubscriber enforces RS rules more strictly when going out of the Reactor world and bridging to an external Publisher (#711)

• The reactor-test artifact is now part of the io.projectreactor groupId (and reactor-core github repository)

• Kotlin extensions for core and test are now part of the reactor-core repo and artifacts

• behavioral changes

  • doAfterTerminate doesn't cause an onError signal if its callback fails (#270)
  • sample will now include the last element in the source sequence if it terminates with an open sampling window (#724)
  • Flux.cache(Duration) now expires the termination signal as well (#750)

• Flux/Mono method changes

  • ParallelFlux#subscribe(lambdas) lambda-based variants now return a Disposable (composite of all rails subscribers, #800)
  • Flux#toIterable now takes an int for its batchSize argument (f3e13bf)
  • Mono.doOnTerminate and doAfterTerminate now take a simple Runnable (aligning the API with Flux). The old doOnTerminate(BiConsumer<Throwable, T>) can be achieved with the added doOnSuccessOrError and old doAfterTerminate(BiConsumer<Throwable, T>) with doAfterSuccessOrError(BiConsumer<Throwable, T>). (#836 + 80a3210)
  • Flux.firstEmitting is now called first and Flux.firstEmittingWith is now called or, aligning APIs with those of Mono (#849)
  • static Mono.empty(Publisher<?>) has been removed, prefer using the new Mono.when(p) (2e7fdf3)
  • buffer operators that take a (Publisher, Function) as arguments (open boundary and closing selector) have been renamed to bufferWhen. This improves the situation where having multiple lambda-based signature can create ambiguity (even more so in languages like Kotlin) (#542)
  • delay in Flux is removed in favor of Flux.delayElements (see #263)
  • Mono#untilOther removed (use delayUntil(it -> other.take(1)) instead, see #558)

• then:

  • variants of then that took a Supplier have been removed. Use defer if you really need to provide the continuation Mono lazily (#547)
  • Flux.then(Publisher<Void>) has been removed in favor of thenEmpty
  • Mono.then(Function) is now Mono.flatMap

• Mono zip vs when

  • zip is the operator dedicated to producing Tuples (or equivalent combinations) of the elements from multiple sources.
    As a consequence, when and and variants that produced a Tuple have now been renamed zip and zipWith (#789 + 52f7f04)
  • when and and still exist, as operators in the same family as then: they only care about completion signals,
    discarding elements and returning a Mono<Void>.
  • Mono#zip with an iterable: for consistency, swap the arguments (old "function, iterable" version is deprecated, see #619)
  • zip(Iterable) and zip(Function, Iterable)in Flux has been removed (#338, #619)

• ⚠️ Mono.then(Function) and Mono.flatMap(): how to migrate (#516):

  • first replace all usage of Mono.flatMap with flatMapMany
  • then replace all usage of Mono.then(Function) (which has been removed) with flatMap.
  • the gist of this change is that the old flatMap was actually changing the return type to Flux and was thus missing the "Many" suffix we usually use. At the same time, then(Function) was transforming the source value (unlike any other form of then which usually just ignore the emissions and act on the onComplete). Thus it was actually more of a flatMap.

• error handling operators have been made more consistent between Mono and Flux (#533, #535, #531)

  • Flux: onErrorResumeWith -> onErrorResume, mapError -> onErrorMap, switchOnError -> replace with onErrorResume with a lambda that ignores the left hand side.
  • Mono: otherwise -> onErrorResume, otherwiseReturn -> onErrorReturn, otherwiseIfEmpty -> switchIfEmpty, mapError -> onErrorMap

• window:

  • various window operators have been aligned in behavior with respect to cancellation (#384)
  • window operators that take a (Publisher, Function) as arguments (open boundary and closing selector) have been renamed to windowWhen. This improves the situation where having multiple lambda-based signature can create ambiguity (even more so in languages like Kotlin) (#542)
  • Flux.window() has been removed in favor of window(Publisher), windowUntil and windowWhile(see #260)
  • Flux.windowTimeOrSize: the timeout now starts right from subscription, resulting in potentially empty windows being emitted. It also restarts if the
    timeout expires, so there could be several empty windows emitted eg. 3 for a timeout of 100ms if the Flux doesn't emit for 300ms.
    The equivalent buffer version remains unchanged. (#484)
  • windowUntil/windowWhile have Flux windows instead of GroupedFlux (#759)

• Cross operator changes based on prefix / parameter pattern:

  • xxxMillis: Timed operators with the Millis suffix have been removed, use their Duration based counterparts
  • xxx(boolean delayError): All operators that were taking a delayError boolean parameter instead of having a DelayError suffixed variant now use the suffix variant pattern (see #480)
  • xxx(Function<Throwable, ?>): Operators dealing with a Predicate or Function of Throwable have been aligned to take a more generic <? super Throwable>
  • All subscribe methods in Flux that had a prefetch parameter have been removed (you can use limitRate to achieve the same effect)

• class changes

  • Signal is now an interface and MutableNextSignal has been removed (#779)
  • Old introspection interfaces, which were mostly only used internally, have been removed in favor of the single Scannable interface (Loopback, MultiProducer, MultiReceiver, Producer, Receiver, Trackable, see #249, 20bd64d)
  • Cancellation has been removed and replaced with Disposable (see #322, 844769d)
  • TimedScheduler has been removed, as well as the Scheduler#shutdown method (most schedulers are now fully time capable and you can use dispose(). The few Scheduler that are not time-capable will throw a RejectedExecutionException indicating so whenever one attempts to use schedulePeriodically on them. Note that most timed operators now use the parallel() Scheduler by default. (see #451, #322)
  • Review interrupted flag on dispose (#507)
  • Removed fromExecutorService(exec, boolean) variant
  • Removed factory hook for simple ExecutorService (now all Scheduled)
  • OpenHashSet has been removed, as it was only used for Disposables: use Disposables.composite() (40fbd60)
  • QueueSupplier has been renamed to Queues and is now purely about queue-related utils and static suppliers. It doesn't implement Supplier<Queue> itself anymore. (#733)

• Hooks

  • Hooks are now cumulative. Setting a hook will add it on top of the existing one(s). You can also name
    some hooks, which allows to update a sub-hook or only partially reset a hook. (#687, #784)
  • new hooks onEachOperator + onLastOperator, now Function<Publisher, Publisher>, new operator lift (#775)
  • hooks must now be called with the Context (#830)

• nullability

  • Nullability checks are helped with the use of annotations. Most of the Reactor API is NonNull (#614)
  • Some additional non-null checks are enforced, notably on all Tuple (#696)

• Processors have been reworked:

  • You cannot connect explicitly anymore but should rather use the sink() method. For details and rationale, see #525.
  • Processors that are costly to instantiate (TopicProcessor, WorkQueueProcessor) now have a Builder instead of
    factory methods. However, they keep a couple of significant factory methods (see #471, #616 and #628)
  • Mono#subscribe() returns a Disposable. Use .toProcessor() to get the exact legacy behavior (see #566, #605 and #638)
  • UnicastProcessor won't ignore the overflowStrategy if we can detect that a bounded queue is used. By default, an unbounded queue is used and the strategy is silently ignored. (#612)
  • DirectProcessor and UnicastProcessor now call onErrorDropped and onNextDropped if calling after terminate. Exceptions.argumentIsNullException() has been removed (most of the time replaced by Objects.requireNonNull()). (7f6ff86)
  • MonoProcessor cancel/dispose now signals CancellationException (#792)

• Some changes have been made to the way writing custom operators works

  • Operators.SubscriberAdapter is not part of the public API anymore.
  • Operators.addAndGet(AtomicLong, ...) has been removed in favor of the AtomicLongFieldUpdater alternative, itself renamed to addCap (#371)
  • Removed unused and confusing Operators.addAndGet (#371)
  • MonoSource and FluxSource are now package private, superseded by MonoOperator and FluxOperator (to deal with
    new CoreSubscriber)
  • Operators.onNextDropped default behaviour is to log the drop rather than throw a "fail with cancel" exception (#823)

✨ New features and improvements

• ✨ Associating a Context to a reactive sequence ✨ (Flux or Mono) is now possible. This is the major new feature in this milestone. (#117, #210, #337, #447, #704, #705, #723 and various other PRs)
  • Context can be created using factory methods
  • Added CoreSubscriber, a Reactor-specific Subscriber common to all Flux and Mono operators and needed to pass along the Context
  • Flux#subscribe(Subscriber) is now final, and one needs to instead implement subscribe(CoreSubscriber) in custom operators / Flux.
  • The context is enriched during the subscription phase, via the subscriberContext(Function<Context, Context>).
  • There is a static factory method Mono#currentContext that can be used to emit the Context initialized downstream of it (or propagated inside eg. a flatMap)
  • onErrorDropped/onNextDropped/onOperatorError/onRejectedExecution look for local hooks in the Context (b576f7f)
• Flux.error and Mono.error are now ScalarCallable and the error case can be fused when source is scalar (#716)
• Introspection is now doable on Scannable classes via an Attr key. Attributes are quasi-enumerations exposing constants rather than enums. This allows splitting them into typed keys with meaningful global defaults so that scan doesn't need a Class to attempt a cast. Operator implementation is based on a Object scan, scanUnsafe.(see #606, #728)
• Added the expand operator (#769)
• Mono#fromCallable/fromSupplier now map null result to empty Mono (#743)
• Null analysis has been enabled through the use of Nullable/NonNullApi annotations (#614, #864, #873, #875)
• Upgrade to Kotlin 1.1.4-3
• Add ability to name and tag a reactive sequence (#579)
• Add a new Console logger, now the default fallback instead of JDK logger (#680)
• Add utility method to check if a Throwable is composite (#770)
• Add Swap & Composite Disposable specializations (#731, bf8d8cb)
• onEachOperator/onLastOperator/onOperatorError hooks can be named, allowing for additive hooks AND partial hook reset (#784)
• Offer way to do requests on original thread with subscribeOn (#777)
• Add take(Duration) and takeUntilOther(Publisher) to Mono (#797)
• Have ParallelFlux lambda subscribe return composite Disposable (#800)
• Add Disposable single(), disposed() and never() factories to a Disposables utility class ( #799, #804, #812)
• Add Mono.cache(Duration) (#683)
• Add "not dropped" assertions to StepVerifier assertions (96ae5ca)
• Add a PublisherProbe to easily probe for subscription in tests (#833)
• Return original reference if unwrapped exception is null (#848)
• Reduce WorkQueueProcessor/TopicProcessor sink serialization when it is not necessary (#630, #727)
• Added a distinctUntilChanged variant with a bipredicate to evaluate if there is a change (#608)
• Added a onBackpressureBuffer variant with a TTL (#296)
• Added an optional configurable default timeout on StepVerifier#verify() (#651)
• Added expect|verifyErrorSatisfies StepVerifier error expectations (#670)
• When using Mono.fromRunnable, the resulting Mono's generic type is better inferred (#686)
• Assembly tracking internals have been improved, allowing to cache assembly stacktrace notably (#712)
• the checkpoint description is included inFlux/MonoSource#toString (#611)
• checkpoint(String) is now light by default, lowering the cost of using that operator (see #587)
• the Supplier used for distinct can be tuned (#577)
• refCount now has a variant that waits for a grace period before unsubscribing from upstream (#569, #624, #627)
• Added a Flux.then(Mono) operator to align with the Mono equivalent (#547)
• Added Mono.delayUntil to delay the emission of a Mono until after a companion Publisher generated from the source value completes. This acts on onComplete, unlike now removed untilOther. (#558, #568, #674)
• The error message in case some operators badly behave with backpressure ("Queue full?!") has been made more explicit (#540)
• ParallelSubscriber now has a subscribe() method and all subscribe(...) methods are final like in Flux (#564)
• StepVerfier improvements:
  • Added overloads to Step#expectNext for up to 6 parameters (reactor/reactor-addons#106)
  • Valued expectNextSequence should fail when checking empty Flux (reactor/reactor-addons#98)

🪲 Bug fixes

• Fix recursive context bug on flatMapMany (8e33991)
• Add reason to default Scheduler exception throwing (#854)
• Change Operators subscriptions from enum to final classes (#857)
• Safely suppress exceptions through Exceptions.addSuppressed (#758)
• ParallelFlux now correctly support conditional subscribers (#865, #389)
• MonoProcessor shouldn't have been a strict subscriber (#866)
• Review int.max prefetch (f3e13bf)
• Multicast cancel now disposes upstream properly (#870)
• Protect serializedSink from illegal subscriber onNext throw (#613)
• Improve progressive demand handling (#371)
• ParallelScheduler does not schedule task evenly between threads (#761)
• Pass failed signal to onOperatorError in trySubscribeScalarMap (#684)
• Remove cancel during onNext for MonoDelayElement (#749)
• Attr.LARGE_BUFFERED defaults back to null (as in "irrelevant") (#751)
• Revert publishOn async path detection over-optimization (#767)
• Exceptions.addThrowable now reuse same root composite (#771)
• Guard ExecutorScheduler against Executor's task failures (#324)
• Catch fused flattenIterable polling failures (#841)
• Fix windowTimeout rejection double cancel (31ef92d)
• VirtualTimeScheduler shutdown is not correctly handled with periods (#776)
• Protect immediate flatMap scalar emissions from out of order consuming (c7d3c12, 3abcb83, ce49077)
• Ensure no Subscription#request throws/errors (4d61440, decfa13)
• Avoid the shortcut of MonoProcessor.onNext(null) for onComplete (#701)
• StepVerifier now ignores empty multivalue expectations, which could previously lead to bad assertions (#650)
• Fixed a bug were macro-fusion of Flux#then and Mono#then would use arrays of incompatible types, resulting in an ArrayStoreException (#661)
• When a doOnNext callback would fail, when fused it wouldn't be caught by a doOnError (#664)
• filterWhen had a bug in tracking request and produced amount, resulting in under-requesting hangs (#689, #692)
• MonoProcessor could unnecessarily retain references to objects from its source. This is fixed by nulling out the source upon all terminations (#690)
• MonoProcessor#block() had a legacy inner timeout. It will now indefinitely wait for the actual completion. On the other hand, using a negative or 0 timeout in the Duration-based variant will now immediately time out. (#722)
• scan accumulation in a live reduction with a seed would always lag by one. It has been modified so that the seed is sent immediately, which allows the accumulator to run in lockstep with the upstream stage. (#609)
• In Flux.create, setting an onDispose callback on the Sink after its onRequest method was called was ignored. The callback is now explicitly invoked if set after the sink was terminated. (#643)
• a memory leak has been fixed in SingleScheduler.schedule (#578)
• Mono.untilOther would not return a new instance when chained with itself (#515)
• Operators#setOnce was cancelling the wrong subscription

📖 Documentation, Tests and Build

• The javadoc has been reviewed and polished across the board, in style and content. Most notably, a full review of
  Flux and Mono javadoc has been made. It also now outputs UTF-8 html, and a few marble diagrams were added
  (#545, #559, #528, #513, #544, #644, #469, #521, #698, 2cd83c2)
• Some operators' usage restrictions have been better documented in the javadoc (#560,#596, #726)
• A JMH test harness has been added (manual run only)
• Contributing to the documentation is explained and made easier with direct edit links (#801, #808)
• The reference documentation has been completed and polished as well(#516, #535, #555, #580, #853, #838, 0e0b6c9, 89eb394, #862, #404)
• Android compatibility and best effort support is described in README statement (#796)

👍 Thanks to the following contributors that also participated to this release

(and all the intermediate milestones)

@akarnokd, @Buzzardo, @Dmitriusan, @fabriziofortino, @garyrussell, @helmbold, @IlyaZinkovich, @jakekdodd, @jimhorng, @kamilszymanski, @lebannen, @making, @markotron, @olibye, @rajinisivaram, @sdeleuze, @schauder, @TomekJurkowski, @violetagg, @zgagnon