reactor.core.publisher

Class Operators

java.lang.Object

reactor.core.publisher.Operators

public abstract class Operators
extends Object

An 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 http://github.com/reactor/reactive-streams-commons

Nested Class Summary

Nested Classes

Modifier and Type	Class and 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 and Description
static <T> long	addCap(AtomicLongFieldUpdater<T> updater, T instance, long toAdd) Concurrent addition bound to Long.MAX_VALUE.
static long	addCap(long a, long b) Cap an addition to Long.MAX_VALUE
static <T> Fuseable.QueueSubscription<T>	as(Subscription s) Returns the subscription as QueueSubscription if possible or null.
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 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 long	multiplyCap(long a, long b) Cap a multiplication to Long.MAX_VALUE
static void	onErrorDropped(Throwable e, Context context) An unexpected exception is about to be dropped.
static <T> void	onNextDropped(T t, Context context) An unexpected event is about to be dropped.
static Throwable	onOperatorError(Subscription subscription, Throwable error, Context context) Map an "operator" error given an operator parent Subscription.
static Throwable	onOperatorError(Subscription subscription, Throwable error, Object dataSignal, 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, Context context) Return a wrapped RejectedExecutionException which can be thrown by the operator.
static RuntimeException	onRejectedExecution(Throwable original, Subscription subscription, Throwable suppressed, Object dataSignal, 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,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 <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> 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,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,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,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> 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 boolean	validate(long n) Evaluate if a request is strictly positive otherwise reportBadRequest(long)
static boolean	validate(Subscription current, Subscription next) Check Subscription current state and cancel new Subscription if current is push, 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 Detail

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

@Nullable
public static <T> 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

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

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)

Type Parameters:

I - the input type

O - the output type

Parameters:

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.

Returns:

a new Function

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)

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

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.

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

onErrorDropped

public static void onErrorDropped(Throwable e,
                                  Context context)

An unexpected exception is about to be dropped.

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.bubble(Throwable) and onOperatorError(Subscription, Throwable, Object, Context).

Parameters:

original - the original execution error

context - a context that might hold a local error consumer

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,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

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,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,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 push 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,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 push, 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