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	addAndGet(AtomicLongFieldUpdater<T> updater, T instance, long n) 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	checkRequest(long n) Throws an exception if request is 0 or negative as specified in rule 3.09 of Reactive Streams
static boolean	checkRequest(long n, Subscriber<?> subscriber) Propagate an exception to a subscriber if request is 0 or negative, as specified in rule 3.09 of Reactive Streams
static void	complete(Subscriber<?> s) Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onComplete.
static <T> Subscriber<T>	drainSubscriber() Return a singleton Subscriber that does not check for double onSubscribe and purely request Long.MAX.
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 <T> long	getAndAddCap(AtomicLongFieldUpdater<T> updater, T instance, long toAdd) Concurrent addition bound to Long.MAX_VALUE.
static long	multiplyCap(long a, long b) Cap a multiplication to Long.MAX_VALUE
static void	onErrorDropped(Throwable e) An unexpected exception is about to be dropped.
static void	onErrorDropped(Throwable e, Throwable root) An unexpected exception is about to be dropped, and it additionally masks another one due to callback failure.
static <T> void	onNextDropped(T t) An unexpected event is about to be dropped.
static Throwable	onOperatorError(Subscription subscription, Throwable error) Map an "operator" error given an operator parent Subscription.
static Throwable	onOperatorError(Subscription subscription, Throwable error, Object dataSignal) Map an "operator" error given an operator parent Subscription.
static Throwable	onOperatorError(Throwable error) Map an "operator" error.
static RuntimeException	onRejectedExecution() Return a wrapped RejectedExecutionException which can be thrown by the operator.
static RuntimeException	onRejectedExecution(Subscription subscription, Throwable suppressed, Object dataSignal) 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) Throw an IllegalArgumentException if the request is null or negative.
static void	reportMoreProduced() Throw an IllegalStateException that indicates more than the requested amount was produced.
static void	reportSubscriptionSet() Log a duplicate subscription error.
static <T> Subscription	scalarSubscription(Subscriber<? super T> subscriber, T value) Represents a fuseable Subscription that emits a single constant value synchronously to a Subscriber or consumer.
static <T> Subscriber<T>	serialize(Subscriber<? 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 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 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 Detail

addAndGet

public static <T> long addAndGet(AtomicLongFieldUpdater<T> updater,
                                 T instance,
                                 long n)

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

n - delta to add

Returns:

Addition result or Long.MAX_VALUE

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

as

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

checkRequest

public static void checkRequest(long n)
                         throws IllegalArgumentException

Throws an exception if request is 0 or negative as specified in rule 3.09 of Reactive Streams

Parameters:

n - demand to check

Throws:

IllegalArgumentException - the nullOrNegativeRequestException instance

checkRequest

public static boolean checkRequest(long n,
                                   Subscriber<?> subscriber)

Propagate an exception to a subscriber if request is 0 or negative, as specified in rule 3.09 of Reactive Streams

Parameters:

n - demand to check

subscriber - Subscriber to onError if non strict positive n

Returns:

true if valid or false if specification exception occurred

Throws:

IllegalArgumentException - if subscriber is null and demand is negative or 0.

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

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

getAndAddCap

public static <T> long getAndAddCap(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

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,
                                  Throwable root)

An unexpected exception is about to be dropped, and it additionally masks another one due to callback failure. The later will be suppressed by the dropped exception.

Parameters:

e - the exception to handle

root - the optional root cause to suppress

onErrorDropped

public static void onErrorDropped(Throwable e)

An unexpected exception is about to be dropped.

Parameters:

e - the dropped exception

onNextDropped

public static <T> void onNextDropped(T t)

An unexpected event is about to be dropped.

Type Parameters:

T - the dropped value type

Parameters:

t - the dropped data

onOperatorError

public static Throwable onOperatorError(Throwable error)

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

Returns:

mapped Throwable

onOperatorError

public static Throwable onOperatorError(Subscription subscription,
                                        Throwable error)

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

Returns:

mapped Throwable

onOperatorError

public static Throwable onOperatorError(Subscription subscription,
                                        Throwable error,
                                        Object dataSignal)

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

Returns:

mapped Throwable

onRejectedExecution

public static RuntimeException onRejectedExecution()

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).

onRejectedExecution

public static RuntimeException onRejectedExecution(Subscription subscription,
                                                   Throwable suppressed,
                                                   Object dataSignal)

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) (with the passed Subscription).

Parameters:

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) (or null if not relevant)

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)

Throw an IllegalArgumentException if the request is null or negative.

Parameters:

n - the demand to evaluate

See Also:

Exceptions.nullOrNegativeRequestException(long)

reportMoreProduced

public static void reportMoreProduced()

Throw 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.

scalarSubscription

public static <T> Subscription scalarSubscription(Subscriber<? 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> Subscriber<T> serialize(Subscriber<? 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 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,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(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