Package reactor.core.publisher

Class Operators

java.lang.Object

reactor.core.publisher.Operators

public abstract class Operators
extends Object

A helper to support "Operator" writing, handle noop subscriptions, validate request size and to cap concurrent additive operations to Long.MAX_VALUE, which is generic to Subscription.request(long) handling. Combine utils available to operator implementations, @see https://github.com/reactor/reactive-streams-commons

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	Operators.DeferredSubscription	Base class for Subscribers that will receive their Subscriptions at any time, yet they might also need to be cancelled or requested at any time.
static class	Operators.MonoSubscriber<I,O>	A Subscriber/Subscription barrier that holds a single value at most and properly gates asynchronous behaviors resulting from concurrent request or cancel and onXXX signals.

Method Summary

Modifier and Type	Method	Description
static long	addCap(long a, long b)	Cap an addition to Long.MAX_VALUE
static <T> long	addCap(AtomicLongFieldUpdater<T> updater, T instance, long toAdd)	Concurrent addition bound to Long.MAX_VALUE.
static <T> @Nullable Fuseable.QueueSubscription<T>	as(Subscription s)	Returns the subscription as QueueSubscription if possible or null.
static boolean	canAppearAfterOnSubscribe(Subscription subscription)	Check whether the provided Subscription is the one used to satisfy Spec's §1.9 rule before signalling an error.
static Subscription	cancelledSubscription()	A singleton Subscription that represents a cancelled subscription instance and should not be leaked to clients as it represents a terminal state.
static void	complete(Subscriber<?> s)	Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onComplete.
static <T> CoreSubscriber<T>	drainSubscriber()	Return a singleton Subscriber that does not check for double onSubscribe and purely request Long.MAX.
static <T> CoreSubscriber<T>	emptySubscriber()	A Subscriber that is expected to be used as a placeholder and never actually be called.
static Subscription	emptySubscription()	A singleton enumeration that represents a no-op Subscription instance that can be freely given out to clients.
static final Context	enableOnDiscard(@Nullable Context target, Consumer<?> discardConsumer)	Utility method to activate the onDiscard feature (see Flux.doOnDiscard(Class, Consumer)) in a target Context.
static void	error(Subscriber<?> s, Throwable e)	Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onError with the supplied error.
static <I, O> Function<? super Publisher<I>,? extends Publisher<O>>	lift(BiFunction<Scannable,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super I>> lifter)	Create a function that can be used to support a custom operator via CoreSubscriber decoration.
static <O> Function<? super Publisher<O>,? extends Publisher<O>>	lift(Predicate<Scannable> filter, BiFunction<Scannable,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super O>> lifter)	Create a function that can be used to support a custom operator via CoreSubscriber decoration.
static <I, O> Function<? super Publisher<I>,? extends Publisher<O>>	liftPublisher(BiFunction<Publisher,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super I>> lifter)	Create a function that can be used to support a custom operator via CoreSubscriber decoration.
static <O> Function<? super Publisher<O>,? extends Publisher<O>>	liftPublisher(Predicate<Publisher> filter, BiFunction<Publisher,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super O>> lifter)	Create a function that can be used to support a custom operator via CoreSubscriber decoration.
static long	multiplyCap(long a, long b)	Cap a multiplication to Long.MAX_VALUE
static <T> void	onDiscard(@Nullable T element, Context context)	Invoke a (local or global) hook that processes elements that get discarded.
static void	onDiscardMultiple(@Nullable Collection<?> multiple, Context context)	Invoke a (local or global) hook that processes elements that get discarded en masse.
static void	onDiscardMultiple(@Nullable Iterator<?> multiple, boolean knownToBeFinite, Context context)	Invoke a (local or global) hook that processes elements that remains in an Iterator.
static void	onDiscardMultiple(@Nullable Spliterator<?> multiple, boolean knownToBeFinite, Context context)	Invoke a (local or global) hook that processes elements that remains in an Spliterator.
static void	onDiscardMultiple(Stream<?> multiple, Context context)	Invoke a (local or global) hook that processes elements that get discarded en masse.
static <T> void	onDiscardQueueWithClear(@Nullable Queue<T> queue, Context context, @Nullable Function<T,Stream<?>> extract)	Invoke a (local or global) hook that processes elements that get discarded en masse after having been enqueued, due to cancellation or error.
static void	onErrorDropped(Throwable e, Context context)	An unexpected exception is about to be dropped.
static <T> CorePublisher<T>	onLastAssembly(CorePublisher<T> source)	Applies the hooks registered with Hooks.onLastOperator(java.util.function.Function<? super org.reactivestreams.Publisher<java.lang.Object>, ? extends org.reactivestreams.Publisher<java.lang.Object>>) and returns CorePublisher ready to be subscribed on.
static <T> void	onNextDropped(T t, Context context)	An unexpected event is about to be dropped.
static <T> @Nullable Throwable	onNextError(@Nullable T value, Throwable error, Context context)	Find the OnNextFailureStrategy to apply to the calling async operator (which could be a local error mode defined in the Context) and apply it.
static <T> @Nullable Throwable	onNextError(@Nullable T value, Throwable error, Context context, Subscription subscriptionForCancel)	Find the OnNextFailureStrategy to apply to the calling operator (which could be a local error mode defined in the Context) and apply it.
static final BiFunction<? super Throwable,Object,? extends Throwable>	onNextErrorFunction(Context context)
static <T> @Nullable Throwable	onNextInnerError(Throwable error, Context context, @Nullable Subscription subscriptionForCancel)	Find the OnNextFailureStrategy to apply to the calling operator (which could be a local error mode defined in the Context) and apply it.
static <T> @Nullable RuntimeException	onNextPollError(@Nullable T value, Throwable error, Context context)	Find the OnNextFailureStrategy to apply to the calling async operator (which could be a local error mode defined in the Context) and apply it.
static Throwable	onOperatorError(@Nullable Subscription subscription, Throwable error, @Nullable Object dataSignal, Context context)	Map an "operator" error given an operator parent Subscription.
static Throwable	onOperatorError(@Nullable Subscription subscription, Throwable error, Context context)	Map an "operator" error given an operator parent Subscription.
static Throwable	onOperatorError(Throwable error, Context context)	Map an "operator" error.
static RuntimeException	onRejectedExecution(Throwable original, @Nullable Subscription subscription, @Nullable Throwable suppressed, @Nullable Object dataSignal, Context context)	Return a wrapped RejectedExecutionException which can be thrown by the operator.
static RuntimeException	onRejectedExecution(Throwable original, Context context)	Return a wrapped RejectedExecutionException which can be thrown by the operator.
static <T> long	produced(AtomicLongFieldUpdater<T> updater, T instance, long toSub)	Concurrent subtraction bound to 0, mostly used to decrement a request tracker by the amount produced by the operator.
static <F> boolean	replace(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field, F instance, Subscription s)	A generic utility to atomically replace a subscription or cancel the replacement if the current subscription is marked as already cancelled (as in cancelledSubscription()).
static void	reportBadRequest(long n)	Log an IllegalArgumentException if the request is null or negative.
static void	reportMoreProduced()	Log an IllegalStateException that indicates more than the requested amount was produced.
static void	reportSubscriptionSet()	Log a duplicate subscription error.
static void	reportThrowInSubscribe(CoreSubscriber<?> subscriber, Throwable e)	Report a Throwable that was thrown from a call to Publisher.subscribe(Subscriber), attempting to notify the Subscriber by: providing a special Subscription via Subscriber.onSubscribe(Subscription) immediately delivering an onError signal after that
static <T> Subscription	scalarSubscription(CoreSubscriber<? super T> subscriber, T value)	Represents a fuseable Subscription that emits a single constant value synchronously to a Subscriber or consumer.
static <T> Subscription	scalarSubscription(CoreSubscriber<? super T> subscriber, T value, String stepName)	Represents a fuseable Subscription that emits a single constant value synchronously to a Subscriber or consumer.
static <T> CoreSubscriber<T>	serialize(CoreSubscriber<? super T> subscriber)	Safely gate a Subscriber by making sure onNext signals are delivered sequentially (serialized).
static <F> boolean	set(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field, F instance, Subscription s)	A generic utility to atomically replace a subscription or cancel the replacement if current subscription is marked as cancelled (as in cancelledSubscription()) or was concurrently updated before.
static <F> boolean	setOnce(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field, F instance, Subscription s)	Sets the given subscription once and returns true if successful, false if the field has a subscription already or has been cancelled.
static long	subOrZero(long a, long b)	Cap a subtraction to 0
static <F> boolean	terminate(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field, F instance)	Atomically terminates the subscription if it is not already a cancelledSubscription(), cancelling the subscription and setting the field to the singleton cancelledSubscription().
static <T> Fuseable.ConditionalSubscriber<? super T>	toConditionalSubscriber(CoreSubscriber<? super T> actual)	If the actual CoreSubscriber is not Fuseable.ConditionalSubscriber, it will apply an adapter which directly maps all Fuseable.ConditionalSubscriber.tryOnNext(Object) to Subscriber.onNext(Object) and always returns true as the result
static <T> CoreSubscriber<? super T>	toCoreSubscriber(Subscriber<? super T> actual)	If the actual Subscriber is not a CoreSubscriber, it will apply safe strict wrapping to apply all reactive streams rules including the ones relaxed by internal operators based on CoreSubscriber.
static <T> void	toFluxOrMono(Publisher<? extends T>[] sources)
static <T> CorePublisher<T>	toFluxOrMono(Publisher<T> publisher)
static boolean	validate(long n)	Evaluate if a request is strictly positive otherwise reportBadRequest(long)
static boolean	validate(@Nullable Subscription current, Subscription next)	Check Subscription current state and cancel new Subscription if current is set, or return true if ready to subscribe.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details

addCap

public static long addCap(long a,
 long b)

Cap an addition to Long.MAX_VALUE

Parameters:

a - left operand

b - right operand

Returns:

Addition result or Long.MAX_VALUE if overflow

addCap

public static <T> long addCap(AtomicLongFieldUpdater<T> updater,
 T instance,
 long toAdd)

Concurrent addition bound to Long.MAX_VALUE. Any concurrent write will "happen before" this operation.

Type Parameters:

T - the parent instance type

Parameters:

updater - current field updater

instance - current instance to update

toAdd - delta to add

Returns:

value before addition or Long.MAX_VALUE

as

public static <T>
@Nullable Fuseable.QueueSubscription<T> as(Subscription s)

Returns the subscription as QueueSubscription if possible or null.

Type Parameters:

T - the value type of the QueueSubscription.

Parameters:

s - the source subscription to try to convert.

Returns:

the QueueSubscription instance or null

cancelledSubscription

public static Subscription cancelledSubscription()

A singleton Subscription that represents a cancelled subscription instance and should not be leaked to clients as it represents a terminal state.
If algorithms need to hand out a subscription, replace this with a singleton subscription because there is no standard way to tell if a Subscription is cancelled or not otherwise.

Returns:

a singleton noop Subscription to be used as an inner representation of the cancelled state

complete

public static void complete(Subscriber<?> s)

Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onComplete.

Parameters:

s - the target subscriber

drainSubscriber

public static <T> CoreSubscriber<T> drainSubscriber()

Return a singleton Subscriber that does not check for double onSubscribe and purely request Long.MAX. If an error is received it will raise a Exceptions.errorCallbackNotImplemented(Throwable) in the receiving thread.

Returns:

a new Subscriber whose sole purpose is to request Long.MAX

emptySubscriber

public static <T> CoreSubscriber<T> emptySubscriber()

A Subscriber that is expected to be used as a placeholder and never actually be called. All methods log an error.

Type Parameters:

T - the type of data (ignored)

Returns:

a placeholder subscriber

emptySubscription

public static Subscription emptySubscription()

A singleton enumeration that represents a no-op Subscription instance that can be freely given out to clients.

The enum also implements Fuseable.QueueSubscription so operators expecting a QueueSubscription from a Fuseable source don't have to double-check their Subscription received in onSubscribe.

Returns:

a singleton noop Subscription

canAppearAfterOnSubscribe

public static boolean canAppearAfterOnSubscribe(Subscription subscription)

Check whether the provided Subscription is the one used to satisfy Spec's §1.9 rule before signalling an error.

Parameters:

subscription - the subscription to test.

Returns:

true if passed subscription is a subscription created in reportThrowInSubscribe(CoreSubscriber, Throwable).

error

public static void error(Subscriber<?> s,
 Throwable e)

Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onError with the supplied error.

Parameters:

s - target Subscriber to error

e - the actual error

reportThrowInSubscribe

public static void reportThrowInSubscribe(CoreSubscriber<?> subscriber,
 Throwable e)

Report a Throwable that was thrown from a call to Publisher.subscribe(Subscriber), attempting to notify the Subscriber by:

1. providing a special Subscription via Subscriber.onSubscribe(Subscription)
2. immediately delivering an onError signal after that

As at that point the subscriber MAY have already been provided with a Subscription, we assume most well formed subscribers will ignore this second Subscription per Reactive Streams rule 1.9. Subscribers that don't usually ignore may recognize this special case and ignore it by checking canAppearAfterOnSubscribe(Subscription).

Note that if the onSubscribe attempt throws, fatal exceptions are thrown. Other exceptions are added as suppressed on the original exception, which is then directly notified as an onError signal (again assuming that such exceptions occur because a Subscription is already set).

Parameters:

subscriber - the Subscriber being subscribed when the error happened

e - the Throwable that was thrown from Publisher.subscribe(Subscriber)

See Also:

• canAppearAfterOnSubscribe(Subscription)

lift

public static <I,
O>
Function<? super Publisher<I>,? extends Publisher<O>> lift(BiFunction<Scannable,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super I>> lifter)

Create a function that can be used to support a custom operator via CoreSubscriber decoration. The function is compatible with Flux.transform(Function), Mono.transform(Function), Hooks.onEachOperator(Function) and Hooks.onLastOperator(Function), but requires that the original Publisher be Scannable.

This variant attempts to expose the Publisher as a Scannable for convenience of introspection. You should however avoid instanceof checks or any other processing that depends on identity of the Publisher, as it might get hidden if Scannable.isScanAvailable() returns false. Use liftPublisher(BiFunction) instead for that kind of use case.

Type Parameters:

I - the input type

O - the output type

Parameters:

lifter - the bifunction taking Scannable from the enclosing publisher (assuming it is compatible) and consuming CoreSubscriber. It must return a receiving CoreSubscriber that will immediately subscribe to the applied Publisher.

Returns:

a new Function

See Also:

• liftPublisher(BiFunction)

lift

public static <O>
Function<? super Publisher<O>,? extends Publisher<O>> lift(Predicate<Scannable> filter,
 BiFunction<Scannable,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super O>> lifter)

Create a function that can be used to support a custom operator via CoreSubscriber decoration. The function is compatible with Flux.transform(Function), Mono.transform(Function), Hooks.onEachOperator(Function) and Hooks.onLastOperator(Function), but requires that the original Publisher be Scannable.

This variant attempts to expose the Publisher as a Scannable for convenience of introspection. You should however avoid instanceof checks or any other processing that depends on identity of the Publisher, as it might get hidden if Scannable.isScanAvailable() returns false. Use liftPublisher(Predicate, BiFunction) instead for that kind of use case.

The function will be invoked only if the passed Predicate matches. Therefore the transformed type O must be the same than the input type since unmatched predicate will return the applied Publisher.

Type Parameters:

O - the input and output type

Parameters:

filter - the predicate to match taking Scannable from the applied publisher to operate on. Assumes original is scan-compatible.

lifter - the bifunction taking Scannable from the enclosing publisher and consuming CoreSubscriber. It must return a receiving CoreSubscriber that will immediately subscribe to the applied Publisher. Assumes the original is scan-compatible.

Returns:

a new Function

See Also:

• liftPublisher(Predicate, BiFunction)

liftPublisher

public static <I,
O>
Function<? super Publisher<I>,? extends Publisher<O>> liftPublisher(BiFunction<Publisher,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super I>> lifter)

Create a function that can be used to support a custom operator via CoreSubscriber decoration. The function is compatible with Flux.transform(Function), Mono.transform(Function), Hooks.onEachOperator(Function) and Hooks.onLastOperator(Function), and works with the raw Publisher as input, which is useful if you need to detect the precise type of the source (eg. instanceof checks to detect Mono, Flux, true Scannable, etc...).

Type Parameters:

I - the input type

O - the output type

Parameters:

lifter - the bifunction taking the raw Publisher and CoreSubscriber. The publisher can be double-checked (including with instanceof, and the function must return a receiving CoreSubscriber that will immediately subscribe to the Publisher.

Returns:

a new Function

liftPublisher

public static <O>
Function<? super Publisher<O>,? extends Publisher<O>> liftPublisher(Predicate<Publisher> filter,
 BiFunction<Publisher,? super CoreSubscriber<? super O>,? extends CoreSubscriber<? super O>> lifter)

Create a function that can be used to support a custom operator via CoreSubscriber decoration. The function is compatible with Flux.transform(Function), Mono.transform(Function), Hooks.onEachOperator(Function) and Hooks.onLastOperator(Function), and works with the raw Publisher as input, which is useful if you need to detect the precise type of the source (eg. instanceof checks to detect Mono, Flux, true Scannable, etc...).

The function will be invoked only if the passed Predicate matches. Therefore the transformed type O must be the same than the input type since unmatched predicate will return the applied Publisher.

Type Parameters:

O - the input and output type

Parameters:

filter - the Predicate that the raw Publisher must pass for the transformation to occur

lifter - the BiFunction taking the raw Publisher and CoreSubscriber. The publisher can be double-checked (including with instanceof, and the function must return a receiving CoreSubscriber that will immediately subscribe to the Publisher.

Returns:

a new Function

multiplyCap

public static long multiplyCap(long a,
 long b)

Cap a multiplication to Long.MAX_VALUE

Parameters:

a - left operand

b - right operand

Returns:

Product result or Long.MAX_VALUE if overflow

enableOnDiscard

public static final Context enableOnDiscard(@Nullable Context target,
 Consumer<?> discardConsumer)

Utility method to activate the onDiscard feature (see Flux.doOnDiscard(Class, Consumer)) in a target Context. Prefer using the Flux API, and reserve this for testing purposes.

Parameters:

target - the original Context

discardConsumer - the consumer that will be used to cleanup discarded elements

Returns:

a new Context that holds (potentially combined) cleanup Consumer

onDiscard

public static <T> void onDiscard(@Nullable T element,
 Context context)

Invoke a (local or global) hook that processes elements that get discarded. This includes elements that are dropped (for malformed sources), but also filtered out (eg. not passing a filter() predicate).

For elements that are buffered or enqueued, but subsequently discarded due to cancellation or error, see onDiscardMultiple(Stream, Context) and onDiscardQueueWithClear(Queue, Context, Function).

Type Parameters:

T - the type of the element

Parameters:

element - the element that is being discarded

context - the context in which to look for a local hook

See Also:

• onDiscardMultiple(Stream, Context)
• onDiscardMultiple(Collection, Context)
• onDiscardQueueWithClear(Queue, Context, Function)

onDiscardQueueWithClear

public static <T> void onDiscardQueueWithClear(@Nullable Queue<T> queue,
 Context context,
 @Nullable Function<T,Stream<?>> extract)

Invoke a (local or global) hook that processes elements that get discarded en masse after having been enqueued, due to cancellation or error. This method also empties the Queue (either by repeated Queue.poll() calls if a hook is defined, or by Collection.clear() as a shortcut if no hook is defined).

Type Parameters:

T - the type of the element

Parameters:

queue - the queue that is being discarded and cleared

context - the context in which to look for a local hook

extract - an optional extractor method for cases where the queue doesn't directly contain the elements to discard

See Also:

• onDiscardMultiple(Stream, Context)
• onDiscardMultiple(Collection, Context)
• onDiscard(Object, Context)

onDiscardMultiple

public static void onDiscardMultiple(Stream<?> multiple,
 Context context)

Invoke a (local or global) hook that processes elements that get discarded en masse. This includes elements that are buffered but subsequently discarded due to cancellation or error.

Parameters:

multiple - the collection of elements to discard (possibly extracted from other collections/arrays/queues)

context - the Context in which to look for local hook

See Also:

• onDiscard(Object, Context)
• onDiscardMultiple(Collection, Context)
• onDiscardQueueWithClear(Queue, Context, Function)

onDiscardMultiple

public static void onDiscardMultiple(@Nullable Collection<?> multiple,
 Context context)

Invoke a (local or global) hook that processes elements that get discarded en masse. This includes elements that are buffered but subsequently discarded due to cancellation or error.

Parameters:

multiple - the collection of elements to discard

context - the Context in which to look for local hook

See Also:

• onDiscard(Object, Context)
• onDiscardMultiple(Stream, Context)
• onDiscardQueueWithClear(Queue, Context, Function)

onDiscardMultiple

public static void onDiscardMultiple(@Nullable Iterator<?> multiple,
 boolean knownToBeFinite,
 Context context)

Invoke a (local or global) hook that processes elements that remains in an Iterator. Since iterators can be infinite, this method requires that you explicitly ensure the iterator is knownToBeFinite. Typically, operating on an Iterable one can get such a guarantee by looking at the Spliterator's Spliterator.getExactSizeIfKnown().

Parameters:

multiple - the Iterator whose remainder to discard

knownToBeFinite - is the caller guaranteeing that the iterator is finite and can be iterated over

context - the Context in which to look for local hook

See Also:

• onDiscard(Object, Context)
• onDiscardMultiple(Collection, Context)
• onDiscardQueueWithClear(Queue, Context, Function)

onDiscardMultiple

public static void onDiscardMultiple(@Nullable Spliterator<?> multiple,
 boolean knownToBeFinite,
 Context context)

Invoke a (local or global) hook that processes elements that remains in an Spliterator. Since spliterators can be infinite, this method requires that you explicitly ensure the spliterator is knownToBeFinite. Typically, one can get such a guarantee by looking at the Spliterator.getExactSizeIfKnown().

Parameters:

multiple - the Spliterator whose remainder to discard

knownToBeFinite - is the caller guaranteeing that the iterator is finite and can be iterated over

context - the Context in which to look for local hook

See Also:

• onDiscard(Object, Context)
• onDiscardMultiple(Collection, Context)
• onDiscardQueueWithClear(Queue, Context, Function)

onErrorDropped

public static void onErrorDropped(Throwable e,
 Context context)

An unexpected exception is about to be dropped.

If no hook is registered for Hooks.onErrorDropped(Consumer), the dropped error is logged at ERROR level.

Parameters:

e - the dropped exception

context - a context that might hold a local error consumer

onNextDropped

public static <T> void onNextDropped(T t,
 Context context)

An unexpected event is about to be dropped.

If no hook is registered for Hooks.onNextDropped(Consumer), the dropped element is just logged at DEBUG level.

Type Parameters:

T - the dropped value type

Parameters:

t - the dropped data

context - a context that might hold a local next consumer

onOperatorError

public static Throwable onOperatorError(Throwable error,
 Context context)

Map an "operator" error. The result error will be passed via onError to the operator downstream after checking for fatal error via Exceptions.throwIfFatal(Throwable).

Parameters:

error - the callback or operator error

context - a context that might hold a local error consumer

Returns:

mapped Throwable

onOperatorError

public static Throwable onOperatorError(@Nullable Subscription subscription,
 Throwable error,
 Context context)

Map an "operator" error given an operator parent Subscription. The result error will be passed via onError to the operator downstream. Subscription will be cancelled after checking for fatal error via Exceptions.throwIfFatal(Throwable).

Parameters:

subscription - the linked operator parent Subscription

error - the callback or operator error

context - a context that might hold a local error consumer

Returns:

mapped Throwable

onOperatorError

public static Throwable onOperatorError(@Nullable Subscription subscription,
 Throwable error,
 @Nullable Object dataSignal,
 Context context)

Map an "operator" error given an operator parent Subscription. The result error will be passed via onError to the operator downstream. Subscription will be cancelled after checking for fatal error via Exceptions.throwIfFatal(Throwable). Takes an additional signal, which can be added as a suppressed exception if it is a Throwable and the default hook is in place.

Parameters:

subscription - the linked operator parent Subscription

error - the callback or operator error

dataSignal - the value (onNext or onError) signal processed during failure

context - a context that might hold a local error consumer

Returns:

mapped Throwable

onRejectedExecution

public static RuntimeException onRejectedExecution(Throwable original,
 Context context)

Return a wrapped RejectedExecutionException which can be thrown by the operator. This exception denotes that an execution was rejected by a Scheduler, notably when it was already disposed.

Wrapping is done by calling both Exceptions.failWithRejected(Throwable) and onOperatorError(Subscription, Throwable, Object, Context).

Parameters:

original - the original execution error

context - a context that might hold a local error consumer

onNextErrorFunction

public static final BiFunction<? super Throwable,Object,? extends Throwable> onNextErrorFunction(Context context)

onNextError

public static <T> @Nullable Throwable onNextError(@Nullable T value,
 Throwable error,
 Context context,
 Subscription subscriptionForCancel)

Find the OnNextFailureStrategy to apply to the calling operator (which could be a local error mode defined in the Context) and apply it. For poll(), prefer onNextPollError(Object, Throwable, Context) as it returns a RuntimeException.

Cancels the Subscription and return a Throwable if errors are fatal for the error mode, in which case the operator should call onError with the returned error. On the contrary, if the error mode allows the sequence to continue, does not cancel the Subscription and returns null.

Typical usage pattern differs depending on the calling method:

• onNext: check for a throwable return value and call Subscriber.onError(Throwable) if not null, otherwise perform a direct request(1) on the upstream.
• tryOnNext: check for a throwable return value and call Subscriber.onError(Throwable) if not null, otherwise return false to indicate value was not consumed and more must be tried.
• any of the above where the error is going to be propagated through onError but the subscription shouldn't be cancelled: use onNextError(Object, Throwable, Context) instead.
• poll (where the error will be thrown): use onNextPollError(Object, Throwable, Context) instead.

Type Parameters:

T - The type of the value causing the error.

Parameters:

value - The onNext value that caused an error. Can be null.

error - The error.

context - The most significant Context in which to look for an OnNextFailureStrategy.

subscriptionForCancel - The mandatory Subscription that should be cancelled if the strategy is terminal. See also onNextError(Object, Throwable, Context) and onNextPollError(Object, Throwable, Context) for alternatives that don't cancel a subscription

Returns:

a Throwable to propagate through onError if the strategy is terminal and cancelled the subscription, null if not.

onNextError

public static <T> @Nullable Throwable onNextError(@Nullable T value,
 Throwable error,
 Context context)

Find the OnNextFailureStrategy to apply to the calling async operator (which could be a local error mode defined in the Context) and apply it.

This variant never cancels a Subscription. It returns a Throwable if the error is fatal for the error mode, in which case the operator should call onError with the returned error. On the contrary, if the error mode allows the sequence to continue, this method returns null.

Type Parameters:

T - The type of the value causing the error.

Parameters:

value - The onNext value that caused an error.

error - The error.

context - The most significant Context in which to look for an OnNextFailureStrategy.

Returns:

a Throwable to propagate through onError if the strategy is terminal, null if not.

See Also:

• onNextError(Object, Throwable, Context, Subscription)

onNextInnerError

public static <T> @Nullable Throwable onNextInnerError(Throwable error,
 Context context,
 @Nullable Subscription subscriptionForCancel)

Find the OnNextFailureStrategy to apply to the calling operator (which could be a local error mode defined in the Context) and apply it.

Type Parameters:

T - The type of the value causing the error.

Parameters:

error - The error.

context - The most significant Context in which to look for an OnNextFailureStrategy.

subscriptionForCancel - The Subscription that should be cancelled if the strategy is terminal. Null to ignore (for poll, use onNextPollError(Object, Throwable, Context) rather than passing null).

Returns:

a Throwable to propagate through onError if the strategy is terminal and cancelled the subscription, null if not.

onNextPollError

public static <T> @Nullable RuntimeException onNextPollError(@Nullable T value,
 Throwable error,
 Context context)

Find the OnNextFailureStrategy to apply to the calling async operator (which could be a local error mode defined in the Context) and apply it.

Returns a RuntimeException if the error is fatal for the error mode, in which case the operator poll should throw the returned error. On the contrary if the error mode allows the sequence to continue, returns null in which case the operator should retry the poll().

Note that this method wraps checked exceptions in order to return a RuntimeException that can be thrown from an arbitrary method. If you don't want to throw the returned exception and this wrapping behavior is undesirable, but you still don't want to cancel a subscription, you can use onNextError(Object, Throwable, Context) instead.

Type Parameters:

T - The type of the value causing the error.

Parameters:

value - The onNext value that caused an error.

error - The error.

context - The most significant Context in which to look for an OnNextFailureStrategy.

Returns:

a RuntimeException to be thrown (eg. within Queue.poll() if the error is terminal in the strategy, null if not.

See Also:

• onNextError(Object, Throwable, Context)

onLastAssembly

public static <T> CorePublisher<T> onLastAssembly(CorePublisher<T> source)

Applies the hooks registered with Hooks.onLastOperator(java.util.function.Function<? super org.reactivestreams.Publisher<java.lang.Object>, ? extends org.reactivestreams.Publisher<java.lang.Object>>) and returns CorePublisher ready to be subscribed on.

Type Parameters:

T - the type of the value.

Parameters:

source - the original CorePublisher.

Returns:

a CorePublisher to subscribe on.

toFluxOrMono

public static <T> CorePublisher<T> toFluxOrMono(Publisher<T> publisher)

toFluxOrMono

public static <T> void toFluxOrMono(Publisher<? extends T>[] sources)

onRejectedExecution

public static RuntimeException onRejectedExecution(Throwable original,
 @Nullable Subscription subscription,
 @Nullable Throwable suppressed,
 @Nullable Object dataSignal,
 Context context)

Return a wrapped RejectedExecutionException which can be thrown by the operator. This exception denotes that an execution was rejected by a Scheduler, notably when it was already disposed.

Wrapping is done by calling both Exceptions.failWithRejected(Throwable) and onOperatorError(Subscription, Throwable, Object, Context) (with the passed Subscription).

Parameters:

original - the original execution error

subscription - the subscription to pass to onOperatorError.

suppressed - a Throwable to be suppressed by the RejectedExecutionException (or null if not relevant)

dataSignal - a value to be passed to onOperatorError(Subscription, Throwable, Object, Context) (or null if not relevant)

context - a context that might hold a local error consumer

produced

public static <T> long produced(AtomicLongFieldUpdater<T> updater,
 T instance,
 long toSub)

Concurrent subtraction bound to 0, mostly used to decrement a request tracker by the amount produced by the operator. Any concurrent write will "happen before" this operation.

Type Parameters:

T - the parent instance type

Parameters:

updater - current field updater

instance - current instance to update

toSub - delta to subtract

Returns:

value after subtraction or zero

replace

public static <F> boolean replace(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field,
 F instance,
 Subscription s)

A generic utility to atomically replace a subscription or cancel the replacement if the current subscription is marked as already cancelled (as in cancelledSubscription()).

Type Parameters:

F - the instance type

Parameters:

field - The Atomic container

instance - the instance reference

s - the subscription

Returns:

true if replaced

reportBadRequest

public static void reportBadRequest(long n)

Log an IllegalArgumentException if the request is null or negative.

Parameters:

n - the failing demand

See Also:

• Exceptions.nullOrNegativeRequestException(long)

reportMoreProduced

public static void reportMoreProduced()

Log an IllegalStateException that indicates more than the requested amount was produced.

See Also:

• Exceptions.failWithOverflow()

reportSubscriptionSet

public static void reportSubscriptionSet()

Log a duplicate subscription error.

See Also:

• Exceptions.duplicateOnSubscribeException()

scalarSubscription

public static <T> Subscription scalarSubscription(CoreSubscriber<? super T> subscriber,
 T value)

Represents a fuseable Subscription that emits a single constant value synchronously to a Subscriber or consumer.

Type Parameters:

T - the value type

Parameters:

subscriber - the delegate Subscriber that will be requesting the value

value - the single value to be emitted

Returns:

a new scalar Subscription

scalarSubscription

public static <T> Subscription scalarSubscription(CoreSubscriber<? super T> subscriber,
 T value,
 String stepName)

Represents a fuseable Subscription that emits a single constant value synchronously to a Subscriber or consumer. Also give the subscription a user-defined stepName for the purpose of Scannable.stepName().

Type Parameters:

T - the value type

Parameters:

subscriber - the delegate Subscriber that will be requesting the value

value - the single value to be emitted

stepName - the String to represent the Subscription in Scannable.stepName()

Returns:

a new scalar Subscription

serialize

public static <T> CoreSubscriber<T> serialize(CoreSubscriber<? super T> subscriber)

Safely gate a Subscriber by making sure onNext signals are delivered sequentially (serialized). Serialization uses thread-stealing and a potentially unbounded queue that might starve a calling thread if races are too important and Subscriber is slower.

Type Parameters:

T - the relayed type

Parameters:

subscriber - the subscriber to serialize

Returns:

a serializing Subscriber

set

public static <F> boolean set(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field,
 F instance,
 Subscription s)

A generic utility to atomically replace a subscription or cancel the replacement if current subscription is marked as cancelled (as in cancelledSubscription()) or was concurrently updated before.

The replaced subscription is itself cancelled.

Type Parameters:

F - the instance type

Parameters:

field - The Atomic container

instance - the instance reference

s - the subscription

Returns:

true if replaced

setOnce

public static <F> boolean setOnce(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field,
 F instance,
 Subscription s)

Sets the given subscription once and returns true if successful, false if the field has a subscription already or has been cancelled.

If the field already has a subscription, it is cancelled and the duplicate subscription is reported (see reportSubscriptionSet()).

Type Parameters:

F - the instance type containing the field

Parameters:

field - the field accessor

instance - the parent instance

s - the subscription to set once

Returns:

true if successful, false if the target was not empty or has been cancelled

subOrZero

public static long subOrZero(long a,
 long b)

Cap a subtraction to 0

Parameters:

a - left operand

b - right operand

Returns:

Subtraction result or 0 if overflow

terminate

public static <F> boolean terminate(AtomicReferenceFieldUpdater<F,@Nullable Subscription> field,
 F instance)

Atomically terminates the subscription if it is not already a cancelledSubscription(), cancelling the subscription and setting the field to the singleton cancelledSubscription().

Type Parameters:

F - the instance type containing the field

Parameters:

field - the field accessor

instance - the parent instance

Returns:

true if terminated, false if the subscription was already terminated

validate

public static boolean validate(@Nullable Subscription current,
 Subscription next)

Check Subscription current state and cancel new Subscription if current is set, or return true if ready to subscribe.

Parameters:

current - current Subscription, expected to be null

next - new Subscription

Returns:

true if Subscription can be used

validate

public static boolean validate(long n)

Evaluate if a request is strictly positive otherwise reportBadRequest(long)

Parameters:

n - the request value

Returns:

true if valid

toCoreSubscriber

public static <T> CoreSubscriber<? super T> toCoreSubscriber(Subscriber<? super T> actual)

If the actual Subscriber is not a CoreSubscriber, it will apply safe strict wrapping to apply all reactive streams rules including the ones relaxed by internal operators based on CoreSubscriber.

Type Parameters:

T - passed subscriber type

Parameters:

actual - the Subscriber to apply hook on

Returns:

an eventually transformed Subscriber

toConditionalSubscriber

public static <T>
Fuseable.ConditionalSubscriber<? super T> toConditionalSubscriber(CoreSubscriber<? super T> actual)

If the actual CoreSubscriber is not Fuseable.ConditionalSubscriber, it will apply an adapter which directly maps all Fuseable.ConditionalSubscriber.tryOnNext(Object) to Subscriber.onNext(Object) and always returns true as the result

Type Parameters:

T - passed subscriber type

Parameters:

actual - the Subscriber to adapt

Returns:

a potentially adapted Fuseable.ConditionalSubscriber