- * A {@link Publisher} can serve multiple {@link Subscriber}s subscribed {@link #subscribe(Subscriber)} dynamically
+ * A {@link Publisher} can serve multiple {@link Subscriber}s subscribed {@link Publisher#subscribe(Subscriber)} dynamically
* at various points in time.
*
- * @param
* If the {@link Publisher} rejects the subscription attempt or otherwise fails it will
- * signal the error via {@link Subscriber#onError}.
+ * signal the error via {@link Subscriber#onError(Throwable)}.
*
* @param s the {@link Subscriber} that will consume signals from this {@link Publisher}
*/
Index: 3rdParty_sources/reactive-streams/org/reactivestreams/Subscriber.java
===================================================================
diff -u -r2d6722d97aad801e2f7db229945ae3dab6ec8576 -rc4ce08dc0aae7d9da822088a3d5710484f6b0402
--- 3rdParty_sources/reactive-streams/org/reactivestreams/Subscriber.java (.../Subscriber.java) (revision 2d6722d97aad801e2f7db229945ae3dab6ec8576)
+++ 3rdParty_sources/reactive-streams/org/reactivestreams/Subscriber.java (.../Subscriber.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
@@ -1,13 +1,6 @@
-/************************************************************************
- * Licensed under Public Domain (CC0) *
- * *
- * To the extent possible under law, the person who associated CC0 with *
- * this code has waived all copyright and related or neighboring *
- * rights to this code. *
- * *
- * You should have received a copy of the CC0 legalcode along with this *
- * work. If not, see
* Demand can be signaled via {@link Subscription#request(long)} whenever the {@link Subscriber} instance is capable of handling more.
*
- * @param
@@ -36,8 +30,7 @@
*
* The {@link Publisher} will send notifications only in response to {@link Subscription#request(long)}.
*
- * @param s
- * {@link Subscription} that allows requesting data via {@link Subscription#request(long)}
+ * @param s the {@link Subscription} that allows requesting data via {@link Subscription#request(long)}
*/
public void onSubscribe(Subscription s);
Index: 3rdParty_sources/reactive-streams/org/reactivestreams/Subscription.java
===================================================================
diff -u -r2d6722d97aad801e2f7db229945ae3dab6ec8576 -rc4ce08dc0aae7d9da822088a3d5710484f6b0402
--- 3rdParty_sources/reactive-streams/org/reactivestreams/Subscription.java (.../Subscription.java) (revision 2d6722d97aad801e2f7db229945ae3dab6ec8576)
+++ 3rdParty_sources/reactive-streams/org/reactivestreams/Subscription.java (.../Subscription.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
@@ -1,13 +1,6 @@
-/************************************************************************
- * Licensed under Public Domain (CC0) *
- * *
- * To the extent possible under law, the person who associated CC0 with *
- * this code has waived all copyright and related or neighboring *
- * rights to this code. *
- * *
- * You should have received a copy of the CC0 legalcode along with this *
- * work. If not, see
* It is used to both signal desire for data and cancel demand (and allow resource cleanup).
- *
*/
public interface Subscription {
+
/**
* No events will be sent by a {@link Publisher} until demand is signaled via this method.
*
Index: 3rdParty_sources/reactor/reactor/adapter/JdkFlowAdapter.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -rc4ce08dc0aae7d9da822088a3d5710484f6b0402
--- 3rdParty_sources/reactor/reactor/adapter/JdkFlowAdapter.java (.../JdkFlowAdapter.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/adapter/JdkFlowAdapter.java (.../JdkFlowAdapter.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
@@ -57,9 +57,9 @@
}
private static class FlowPublisherAsFlux
+ * Unless wrapped explicitly, such exceptions would always be thrown by operators instead of
+ * propagation through onError, potentially interrupting progress of Flux/Mono sequences.
+ * When they occur, the JVM itself is assumed to be in an unrecoverable state, and so is Reactor.
+ *
+ * @see #throwIfFatal(Throwable)
+ * @see #throwIfJvmFatal(Throwable)
+ * @see #isFatal(Throwable)
+ * @param t the {@link Throwable} to check
+ * @return true if the throwable is considered Jvm Fatal
+ */
+ public static boolean isJvmFatal(@Nullable Throwable t) {
+ return t instanceof VirtualMachineError ||
+ t instanceof ThreadDeath ||
+ t instanceof LinkageError;
+ }
+
+ /**
+ * Check if a {@link Throwable} is considered by Reactor as Fatal and would be thrown by
+ * {@link #throwIfFatal(Throwable)}.
+ *
+ * Unless wrapped explicitly, such exceptions would always be thrown by operators instead of
+ * propagation through onError, potentially interrupting progress of Flux/Mono sequences.
+ * When they occur, the assumption is that Reactor is in an unrecoverable state (notably
+ * because the JVM itself might be in an unrecoverable state).
+ *
+ * @see #throwIfFatal(Throwable)
+ * @see #isJvmFatal(Throwable)
+ * @param t the {@link Throwable} to check
+ * @return true if the throwable is considered fatal
+ */
+ public static boolean isFatal(@Nullable Throwable t) {
+ return isFatalButNotJvmFatal(t) || isJvmFatal(t);
+ }
+
+ /**
+ * Internal intermediate test that only detect Fatal but not Jvm Fatal exceptions.
+ */
+ static boolean isFatalButNotJvmFatal(@Nullable Throwable t) {
+ return t instanceof BubblingException || t instanceof ErrorCallbackNotImplemented;
+ }
+
+ /**
* Throws a particular {@code Throwable} only if it belongs to a set of "fatal" error
* varieties. These varieties are as follows:
+ * Tags can only be discovered until no parent can be inspected, which happens either
+ * when the source publisher has been reached or when a non-reactor intermediate operator
+ * is present in the parent chain (i.e. a stage that is not {@link Scannable} for {@link Attr#PARENT}).
*
- * @return the stream of tags for this {@link Scannable} and its parents
+ * @return the stream of tags for this {@link Scannable} and its reachable parents, including duplicates
+ * @see #tagsDeduplicated()
*/
default Stream
+ * Tags can only be discovered until no parent can be inspected, which happens either
+ * when the source publisher has been reached or when a non-reactor intermediate operator
+ * is present in the parent chain (i.e. a stage that is not {@link Scannable} for {@link Attr#PARENT}).
+ *
+ * @return a {@link Map} of deduplicated tags from this {@link Scannable} and its reachable parents
+ * @see #tags()
+ */
+ default Map
+ * Both publisher-to-subscriber events and subscription events are handled. Methods are closer to the side-effect doOnXxx operators
+ * than to {@link Subscriber} and {@link Subscription} methods, in order to avoid misconstruing this for an actual Reactive Streams
+ * implementation. The actual downstream {@link Subscriber} and upstream {@link Subscription} are intentionally not exposed
+ * to avoid any influence on the observed sequence.
+ *
+ * @author Simon Baslé
+ */
+public interface SignalListener
+ * Once the {@link Publisher} has acknowledged with a {@link Subscription}, the {@link #doOnSubscription()}
+ * handler will be invoked before that {@link Subscription} is passed down.
+ *
+ * @see #doOnSubscription()
+ */
+ void doFirst() throws Throwable;
+
+ /**
+ * Handle terminal signals after the signals have been propagated, as the final step.
+ * Only {@link SignalType#ON_COMPLETE}, {@link SignalType#ON_ERROR} or {@link SignalType#CANCEL} can be passed.
+ * This handler is invoked AFTER the terminal signal has been propagated, and if relevant AFTER the {@link #doAfterComplete()}
+ * or {@link #doAfterError(Throwable)} events. If any doOnXxx handler throws, this handler is NOT invoked (see {@link #handleListenerError(Throwable)}
+ * instead).
+ *
+ * @see #handleListenerError(Throwable)
+ */
+ void doFinally(SignalType terminationType) throws Throwable;
+
+ /**
+ * Handle the fact that the upstream {@link Publisher} acknowledged {@link Subscription}.
+ * The {@link Subscription} is intentionally not exposed in order to avoid manipulation by the observer.
+ *
+ * While {@link #doFirst} is invoked right as the downstream {@link Subscriber} is registered,
+ * this method is invoked as the upstream answers back with a {@link Subscription} (and before that
+ * same {@link Subscription} is passed downstream).
+ *
+ * @see #doFirst()
+ */
+ void doOnSubscription() throws Throwable;
+
+ /**
+ * Handle the negotiation of fusion between two {@link reactor.core.Fuseable} operators. As the downstream operator
+ * requests fusion, the upstream answers back with the compatible level of fusion it can handle. This {@code negotiatedFusion}
+ * code is passed to this handler right before it is propagated downstream.
+ *
+ * @param negotiatedFusion the final fusion mode negotiated by the upstream operator in response to a fusion request
+ * from downstream
+ */
+ void doOnFusion(int negotiatedFusion) throws Throwable;
+
+ /**
+ * Handle a new request made by the downstream, exposing the demand.
+ *
+ * This is invoked before the request is propagated upstream.
+ *
+ * @param requested the downstream demand
+ */
+ void doOnRequest(long requested) throws Throwable;
+
+ /**
+ * Handle the downstream cancelling its currently observed {@link Subscription}.
+ *
+ * This handler is invoked before propagating the cancellation upstream, while {@link #doFinally(SignalType)}
+ * is invoked right after the cancellation has been propagated upstream.
+ *
+ * @see #doFinally(SignalType)
+ */
+ void doOnCancel() throws Throwable;
+
+ /**
+ * Handle a new value emission from the source.
+ *
+ * This handler is invoked before propagating the value downstream.
+ *
+ * @param value the emitted value
+ */
+ void doOnNext(T value) throws Throwable;
+
+ /**
+ * Handle graceful onComplete sequence termination.
+ *
+ * This handler is invoked before propagating the completion downstream, while both
+ * {@link #doAfterComplete()} and {@link #doFinally(SignalType)} are invoked after.
+ *
+ * @see #doAfterComplete()
+ * @see #doFinally(SignalType)
+ */
+ void doOnComplete() throws Throwable;
+
+ /**
+ * Handle onError sequence termination.
+ *
+ * This handler is invoked before propagating the error downstream, while both
+ * {@link #doAfterError(Throwable)} and {@link #doFinally(SignalType)} are invoked after.
+ *
+ * @param error the exception that terminated the sequence
+ * @see #doAfterError(Throwable)
+ * @see #doFinally(SignalType)
+ */
+ void doOnError(Throwable error) throws Throwable;
+
+ /**
+ * Handle graceful onComplete sequence termination, after onComplete has been propagated downstream.
+ *
+ * This handler is invoked after propagating the completion downstream, similar to {@link #doFinally(SignalType)}
+ * and unlike {@link #doOnComplete()}.
+ */
+ void doAfterComplete() throws Throwable;
+
+ /**
+ * Handle onError sequence termination after onError has been propagated downstream.
+ *
+ * This handler is invoked after propagating the error downstream, similar to {@link #doFinally(SignalType)}
+ * and unlike {@link #doOnError(Throwable)}.
+ *
+ * @param error the exception that terminated the sequence
+ */
+ void doAfterError(Throwable error) throws Throwable;
+
+ /**
+ * Handle malformed {@link Subscriber#onNext(Object)}, which are onNext happening after the sequence has already terminated
+ * via {@link Subscriber#onComplete()} or {@link Subscriber#onError(Throwable)}.
+ * Note that after this handler is invoked, the value is automatically {@link Operators#onNextDropped(Object, Context) dropped}.
+ *
+ * If this handler fails with an exception, that exception is {@link Operators#onErrorDropped(Throwable, Context) dropped} before the
+ * value is also dropped.
+ *
+ * @param value the value for which an emission was attempted (which will be automatically dropped afterwards)
+ */
+ void doOnMalformedOnNext(T value) throws Throwable;
+
+ /**
+ * Handle malformed {@link Subscriber#onError(Throwable)}, which means the sequence has already terminated
+ * via {@link Subscriber#onComplete()} or {@link Subscriber#onError(Throwable)}.
+ * Note that after this handler is invoked, the exception is automatically {@link Operators#onErrorDropped(Throwable, Context) dropped}.
+ *
+ * If this handler fails with an exception, that exception is {@link Operators#onErrorDropped(Throwable, Context) dropped} before the
+ * original onError exception is also dropped.
+ *
+ * @param error the extraneous exception (which will be automatically dropped afterwards)
+ */
+ void doOnMalformedOnError(Throwable error) throws Throwable;
+
+ /**
+ * Handle malformed {@link Subscriber#onComplete()}, which means the sequence has already terminated
+ * via {@link Subscriber#onComplete()} or {@link Subscriber#onError(Throwable)}.
+ *
+ * If this handler fails with an exception, that exception is {@link Operators#onErrorDropped(Throwable, Context) dropped}.
+ */
+ void doOnMalformedOnComplete() throws Throwable;
+
+ /**
+ * A special handler for exceptions thrown from all the other handlers.
+ * This method MUST return normally, i.e. it MUST NOT throw.
+ * When a {@link SignalListener} handler fails, callers are expected to first invoke this method then to propagate
+ * the {@code listenerError} downstream if that is possible, terminating the original sequence with the listenerError.
+ *
+ * Typically, this special handler is intended for a last chance at processing the error despite the fact that
+ * {@link #doFinally(SignalType)} is not triggered on handler errors. For example, recording the error in a
+ * metrics backend or cleaning up state that would otherwise be cleaned up by {@link #doFinally(SignalType)}.
+ *
+ * @param listenerError the exception thrown from a {@link SignalListener} handler method
+ */
+ void handleListenerError(Throwable listenerError);
+
+ /**
+ * In some cases, the tap operation should alter the {@link Context} exposed by the operator in order to store additional
+ * data. This method is invoked when the tap subscriber is created, which is between the invocation of {@link #doFirst()}
+ * and the invocation of {@link #doOnSubscription()}. Generally, only addition of new keys should be performed on
+ * the downstream original {@link Context}. Extra care should be exercised if any pre-existing key is to be removed
+ * or replaced.
+ *
+ * @param originalContext the original downstream operator's {@link Context}
+ * @return the {@link Context} to use and expose upstream
+ */
+ default Context addToContext(Context originalContext) {
+ return originalContext;
+ }
+}
Index: 3rdParty_sources/reactor/reactor/core/observability/SignalListenerFactory.java
===================================================================
diff -u
--- 3rdParty_sources/reactor/reactor/core/observability/SignalListenerFactory.java (revision 0)
+++ 3rdParty_sources/reactor/reactor/core/observability/SignalListenerFactory.java (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.observability;
+
+import org.reactivestreams.Publisher;
+
+import reactor.util.context.ContextView;
+
+/**
+ * A factory for per-subscription {@link SignalListener}, exposing the ability to generate common state at publisher level
+ * from the source {@link Publisher}.
+ *
+ * Examples of such state include:
+ *
+ * The {@code source} {@link Publisher} is the same as the one that triggered common state creation at assembly time in
+ * {@link #initializePublisherState(Publisher)}). Said common state is passed to this method as well, and so is the
+ * {@link ContextView} for the newly registered {@link reactor.core.CoreSubscriber}.
+ *
+ * @param source the source {@link Publisher} that is being subscribed to
+ * @param listenerContext the {@link ContextView} associated with the new subscriber
+ * @param publisherContext the common state initialized in {@link #initializePublisherState(Publisher)}
+ * @return a stateful {@link SignalListener} observing signals to and from the new subscriber
+ */
+ SignalListener
+ * This variant uses the implicit global {@code ContextRegistry} and captures from all
+ * available {@code ThreadLocalAccessors}. It is the same variant backing {@link Flux#contextCapture()}
+ * and {@link Mono#contextCapture()}.
+ *
+ * @return the {@link Context} augmented with captured entries
+ */
+ static Function Note, this is only applied to {@link FluxTap}, {@link FluxTapFuseable},
+ * {@link MonoTap}, and {@link MonoTapFuseable}. The automatic propagation
+ * variants: {@link FluxTapRestoringThreadLocals} and
+ * {@link MonoTapRestoringThreadLocals} do not use this method.
+ * @param original the original {@link SignalListener} from the user
+ * @param contextSupplier supplies the potentially modified {@link Context} to
+ * restore {@link ThreadLocal} values from
+ * @return potentially wrapped {@link SignalListener} or the original
+ * @param
*
@@ -777,7 +780,7 @@
* available backpressure modes
* @param emitter Consume the {@link FluxSink} provided per-subscriber by Reactor to generate signals.
* @return a {@link Flux}
- * @see #create(Consumer, reactor.core.publisher.FluxSink.OverflowStrategy)
+ * @see #create(Consumer, OverflowStrategy)
*/
public static
- *
+ * While this operator does retrieve at most one value from each source, it only compares values when two or more
+ * sources emit at the same time. In that case it picks the smallest of these competing values and continues doing so
+ * as long as there is demand. It is therefore best suited for asynchronous sources where you do not want to wait
+ * for a value from each source before emitting a value downstream.
+ *
+ *
+ * While this operator does retrieve at most one value from each source, it only compares values when two or more
+ * sources emit at the same time. In that case it picks the smallest of these competing values and continues doing so
+ * as long as there is demand. It is therefore best suited for asynchronous sources where you do not want to wait
+ * for a value from each source before emitting a value downstream.
+ *
+ *
+ * While this operator does retrieve at most one value from each source, it only compares values when two or more
+ * sources emit at the same time. In that case it picks the smallest of these competing values and continues doing so
+ * as long as there is demand. It is therefore best suited for asynchronous sources where you do not want to wait
+ * for a value from each source before emitting a value downstream.
+ *
+ *
+ * While this operator does retrieve at most one value from each source, it only compares values when two or more
+ * sources emit at the same time. In that case it picks the smallest of these competing values and continues doing so
+ * as long as there is demand. It is therefore best suited for asynchronous sources where you do not want to wait
+ * for a value from each source before emitting a value downstream.
+ *
+ * Note that it is delaying errors until all data is consumed.
+ *
+ *
@@ -1584,7 +1681,7 @@
if (sources.length == 1) {
return from(sources[0]);
}
- return onAssembly(new FluxMergeComparing<>(prefetch, comparator, false, sources));
+ return onAssembly(new FluxMergeComparing<>(prefetch, comparator, false, true, sources));
}
/**
@@ -1616,7 +1713,7 @@
if (sources.length == 1) {
return from(sources[0]);
}
- return onAssembly(new FluxMergeComparing<>(prefetch, comparator, true, sources));
+ return onAssembly(new FluxMergeComparing<>(prefetch, comparator, true, true, sources));
}
/**
@@ -1706,7 +1803,7 @@
if (sources.length == 1) {
return from(sources[0]);
}
- return onAssembly(new FluxMergeComparing<>(prefetch, comparator, true, sources));
+ return onAssembly(new FluxMergeComparing<>(prefetch, comparator, true, true, sources));
}
/**
@@ -1919,26 +2016,31 @@
/**
* Creates a {@link Flux} that mirrors the most recently emitted {@link Publisher},
- * forwarding its data until a new {@link Publisher} comes in in the source.
+ * forwarding its data until a new {@link Publisher} comes in the source.
*
* The resulting {@link Flux} will complete once there are no new {@link Publisher} in
* the source (source has completed) and the last mirrored {@link Publisher} has also
* completed.
*
*
+ * This operator requests the {@code mergedPublishers} source for an unbounded amount of inner publishers,
+ * but doesn't request each inner {@link Publisher} unless the downstream has made
+ * a corresponding request (no prefetch on publishers emitted by {@code mergedPublishers}).
*
* @param mergedPublishers The {@link Publisher} of {@link Publisher} to switch on and mirror.
* @param
* The resulting {@link Flux} will complete once there are no new {@link Publisher} in
* the source (source has completed) and the last mirrored {@link Publisher} has also
@@ -1950,7 +2052,7 @@
* @param prefetch the inner source request size
* @param
+ * Discard Support: This operator discards the currently open buffer upon cancellation or error triggered by a data signal.
+ *
+ * @param maxSize the max collected size
+ * @param maxTime the timeout enforcing the release of a partial buffer
+ * @param fairBackpressure If {@code true}, prefetches {@code maxSize * 4} from upstream and replenishes the buffer when the downstream demand is satisfactory.
+ * When {@code false}, no prefetching takes place and a single buffer is always ready to be pushed downstream.
+ *
+ * @return a microbatched {@link Flux} of {@link List} delimited by given size or a given period timeout
+ */
+ public final Flux
+ * Discard Support: This operator discards the currently open buffer upon cancellation or error triggered by a data signal.
+ *
+ * @param maxSize the max collected size
+ * @param maxTime the timeout enforcing the release of a partial buffer
+ * @param timer a time-capable {@link Scheduler} instance to run on
+ * @param fairBackpressure If {@code true}, prefetches {@code maxSize * 4} from upstream and replenishes the buffer when the downstream demand is satisfactory.
+ * When {@code false}, no prefetching takes place and a single buffer is always ready to be pushed downstream.
+ *
+ * @return a microbatched {@link Flux} of {@link List} delimited by given size or a given period timeout
+ */
+ public final Flux
+ * Discard Support: This operator discards the currently open buffer upon cancellation or error triggered by a data signal.
+ *
+ * @param maxSize the max collected size
+ * @param maxTime the timeout enforcing the release of a partial buffer
+ * @param bufferSupplier a {@link Supplier} of the concrete {@link Collection} to use for each buffer
+ * @param
+ * Discard Support: This operator discards the currently open buffer upon cancellation or error triggered by a data signal.
+ *
+ * @param maxSize the max collected size
+ * @param maxTime the timeout enforcing the release of a partial buffer
+ * @param timer a time-capable {@link Scheduler} instance to run on
+ * @param bufferSupplier a {@link Supplier} of the concrete {@link Collection} to use for each buffer
+ * @param
* It should be placed towards the end of the reactive chain, as errors
- * triggered downstream of it cannot be observed and augmented with the backtrace.
+ * triggered downstream of it cannot be observed and augmented with the traceback.
*
* The traceback is attached to the error as a {@link Throwable#getSuppressed() suppressed exception}.
* As such, if the error is a {@link Exceptions#isMultiple(Throwable) composite one}, the traceback
@@ -3454,7 +3658,7 @@
*
* Discard Support: This operator discards the intermediate container (see {@link Collector#supplier()} upon
+ * Discard Support: This operator discards the intermediate container (see {@link Collector#supplier()}) upon
* cancellation, error or exception while applying the {@link Collector#finisher()}. Either the container type
* is a {@link Collection} (in which case individual elements are discarded) or not (in which case the entire
* container is discarded). In case the accumulator {@link BiConsumer} of the collector fails to accumulate
@@ -3733,7 +3937,6 @@
*
* @return a {@link Mono} of a sorted {@link List} of all values from this {@link Flux}
*/
- @SuppressWarnings({ "unchecked", "rawtypes" })
public final Mono
* Errors will immediately short circuit current concat backlog.
+ * Note that no prefetching is done on the source, which gets requested only if there
+ * is downstream demand.
*
*
*
* Errors will immediately short circuit current concat backlog. The prefetch argument
- * allows to give an arbitrary prefetch size to the upstream source.
+ * allows to give an arbitrary prefetch size to the upstream source, or to disable
+ * prefetching with {@code 0}.
*
*
* Discard Support: This operator discards elements it internally queued for backpressure upon cancellation.
*
* @param mapper the function to transform this sequence of T into concatenated sequences of V
- * @param prefetch the number of values to prefetch from upstream source (set it to 0 if you don't want it to prefetch)
+ * @param prefetch the number of values to prefetch from upstream source, or {@code 0} to disable prefetching
* @param
* Errors in the individual publishers will be delayed at the end of the whole concat
- * sequence (possibly getting combined into a {@link Exceptions#isMultiple(Throwable) composite}
+ * sequence (possibly getting combined into a {@link Exceptions#isMultiple(Throwable) composite})
* if several sources error.
+ * Note that no prefetching is done on the source, which gets requested only if there
+ * is downstream demand.
*
*
*
* Errors in the individual publishers will be delayed at the end of the whole concat
- * sequence (possibly getting combined into a {@link Exceptions#isMultiple(Throwable) composite}
+ * sequence (possibly getting combined into a {@link Exceptions#isMultiple(Throwable) composite})
* if several sources error.
- * The prefetch argument allows to give an arbitrary prefetch size to the upstream source.
+ * The prefetch argument allows to give an arbitrary prefetch size to the upstream source,
+ * or to disable prefetching with {@code 0}.
*
*
* Discard Support: This operator discards elements it internally queued for backpressure upon cancellation.
*
* @param mapper the function to transform this sequence of T into concatenated sequences of V
- * @param prefetch the number of values to prefetch from upstream source
+ * @param prefetch the number of values to prefetch from upstream source, or {@code 0} to disable prefetching
* @param
* Errors in the individual publishers will be delayed after the current concat
* backlog if delayUntilEnd is false or after all sources if delayUntilEnd is true.
- * The prefetch argument allows to give an arbitrary prefetch size to the upstream source.
+ * The prefetch argument allows to give an arbitrary prefetch size to the upstream source,
+ * or to disable prefetching with {@code 0}.
*
*
*
+ * As a result this operator should generally be used as close as possible to the end of
+ * the chain / subscription point.
+ *
+ * If the {@link ContextView} visible upstream is not empty, a small subset of operators will automatically
+ * restore the context snapshot ({@link #handle(BiConsumer) handle}, {@link #tap(SignalListenerFactory) tap}).
+ * If context-propagation is not available at runtime, this operator simply returns the current {@link Flux}
+ * instance.
+ *
+ * @return a new {@link Flux} where context-propagation API has been used to capture entries and
+ * inject them into the {@link Context}
+ * @see #handle(BiConsumer)
+ * @see #tap(SignalListenerFactory)
+ */
+ public final Flux
@@ -4170,7 +4411,7 @@
*
* @param delay {@link Duration} to shift the sequence by
*
- * @return an shifted {@link Flux} emitting at the same frequency as the source
+ * @return a shifted {@link Flux} emitting at the same frequency as the source
*/
public final Flux
* With this operator, a source emitting at 10Hz with a delaySequence {@link Duration}
@@ -4202,7 +4443,7 @@
* @param delay {@link Duration} to shift the sequence by
* @param timer a time-capable {@link Scheduler} instance to delay signals on
*
- * @return an shifted {@link Flux} emitting at the same frequency as the source
+ * @return a shifted {@link Flux} emitting at the same frequency as the source
*/
public final Flux
* That is: emit the values from this {@link Flux} first, then expand each at a first level of
* recursion and emit all of the resulting values, then expand all of these at a second
- * level and so on..
+ * level and so on.
*
* For example, given the hierarchical structure
* Error Mode Support: This operator supports {@link #onErrorContinue(BiConsumer) resuming on errors} (including when
* fusion is enabled) when the {@link BiConsumer} throws an exception or if an error is signaled explicitly via
* {@link SynchronousSink#error(Throwable)}.
+ *
+ * When the context-propagation library
+ * is available at runtime and the downstream {@link ContextView} is not empty, this operator implicitly uses the
+ * library to restore thread locals around the handler {@link BiConsumer}. Typically, this would be done in conjunction
+ * with the use of {@link #contextCapture()} operator down the chain.
*
* @param handler the handling {@link BiConsumer}
- *
* @param
* Equivalent to {@code flux.publishOn(Schedulers.immediate(), prefetchRate).subscribe() }.
@@ -6028,7 +6268,7 @@
*
* @return a {@link Flux} limiting downstream's backpressure
* @see #publishOn(Scheduler, int)
- * @see #limitRequest(long)
+ * @see #take(long)
*/
public final Flux
* Similar to {@code flux.publishOn(Schedulers.immediate(), prefetchRate).subscribe() },
@@ -6070,7 +6310,7 @@
* @return a {@link Flux} limiting downstream's backpressure and customizing the
* replenishment request amount
* @see #publishOn(Scheduler, int)
- * @see #limitRequest(long)
+ * @see #take(long)
*/
public final Flux
* The {@link MeterRegistry} used by reactor can be configured via
- * {@link reactor.util.Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
+ * {@link Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
* prior to using this operator, the default being
* {@link io.micrometer.core.instrument.Metrics#globalRegistry}.
*
* @return an instrumented {@link Flux}
*
* @see #name(String)
* @see #tag(String, String)
+ * @deprecated Prefer using the {@link #tap(SignalListenerFactory)} with the {@link SignalListenerFactory} provided by
+ * the new reactor-core-micrometer module. To be removed in 3.6.0 at the earliest.
*/
+ @Deprecated
public final Flux
- * If {@link #metrics()} operator is called later in the chain, this name will be used as a prefix for meters' name.
+ * The name is typically visible at assembly time by the {@link #tap(SignalListenerFactory)} operator,
+ * which could for example be configured with a metrics listener using the name as a prefix for meters' id.
*
* @param name a name for the sequence
*
@@ -6486,7 +6730,7 @@
/**
* Evaluate each accepted value against the given {@link Class} type. If the
- * a value matches the type, it is passed into the resulting {@link Flux}. Otherwise
+ * value matches the type, it is passed into the resulting {@link Flux}. Otherwise
* the value is ignored and a request of 1 is emitted.
*
*
@@ -6718,7 +6962,7 @@
* Discard Support: This operator discards elements that it drops after having passed
* them to the provided {@code onDropped} handler.
*
- * @param onDropped the Consumer called when an value gets dropped due to lack of downstream requests
+ * @param onDropped the Consumer called when a value gets dropped due to lack of downstream requests
* @return a backpressured {@link Flux} that drops overflowing elements
*/
public final Flux
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
*
+ *
+ *
+ * @param {@code while (sink.tryEmitNext(v).hasFailed()) {
* LockSupport.parkNanos(10);
* }
* }
*/
@Deprecated
-public final class EmitterProcessor
* 
- *
- * @param contextualPublisherFactory the {@link Publisher} {@link Function} to call on subscribe
- * @param 
+ *
+ * @param sources {@link Publisher} sources of {@link Comparable} to merge
+ * @param a {@link Comparable} merged type that has a {@link Comparator#naturalOrder() natural order}
+ * @return a merged {@link Flux} that compares the latest available value from each source, publishing the
+ * smallest value and replenishing the source that produced it.
+ */
+ @SafeVarargs
+ public static > Flux mergePriority(Publisher... sources) {
+ return mergePriority(Queues.SMALL_BUFFER_SIZE, Comparator.naturalOrder(), sources);
+ }
+
+ /**
+ * Merge data from provided {@link Publisher} sequences into an ordered merged sequence,
+ * by picking the smallest values from each source (as defined by the provided
+ * {@link Comparator}) as they arrive. This is not a {@link #sort(Comparator)}, as it doesn't consider
+ * the whole of each sequences. Unlike mergeComparing, this operator does not wait for a value from each
+ * source to arrive either.
+ * 
+ *
+ * @param comparator the {@link Comparator} to use to find the smallest value
+ * @param sources {@link Publisher} sources to merge
+ * @param
+ *
+ * @param prefetch the number of elements to prefetch from each source (avoiding too
+ * many small requests to the source when picking)
+ * @param comparator the {@link Comparator} to use to find the smallest value
+ * @param sources {@link Publisher} sources to merge
+ * @param
+ *
+ * @param prefetch the number of elements to prefetch from each source (avoiding too
+ * many small requests to the source when picking)
+ * @param comparator the {@link Comparator} to use to find the smallest value
+ * @param sources {@link Publisher} sources to merge
+ * @param 
+ * 
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ * 
*
- * 
@@ -3771,9 +3976,8 @@
*
* @return a concatenated {@link Flux}
*/
- public final
*
*
@@ -3851,7 +4057,7 @@
*
*/
public final
*
*
@@ -3924,14 +4131,14 @@
* @param mapper the function to transform this sequence of T into concatenated sequences of V
* @param delayUntilEnd delay error until all sources have been consumed instead of
* after the current source
- * @param prefetch the number of values to prefetch from upstream source
+ * @param prefetch the number of values to prefetch from upstream source, or {@code 0} to disable prefetching
* @param
@@ -4996,7 +5234,7 @@
* @param capacityHint a capacity hint to prepare the inner queues to accommodate n
* elements per level of recursion.
*
- * @return an breadth-first expanded {@link Flux}
+ * @return a breadth-first expanded {@link Flux}
*/
public final Flux
+ *
+ * @return a new {@link Flux} falling back on completion when an onError occurs
+ * @see #onErrorReturn(Object)
+ */
+ public final Flux
+ *
+ * @return a new {@link Flux} falling back on completion when a matching error occurs
+ * @see #onErrorReturn(Class, Object)
+ */
+ public final Flux
+ *
+ * @return a new {@link Flux} falling back on completion when a matching error occurs
+ * @see #onErrorReturn(Predicate, Object)
+ */
+ public final Flux