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 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.
static class	Operators.SubscriberAdapter<I,O> A Subscriber with an asymetric typed wrapped subscriber.

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	addAndGet(AtomicLong current, long toAdd) Deprecated. This util is neither used by Core nor consistent with the other `addAndGet` methods present which use AtomicLongFieldUpdater. In an effort to keep Operators API consistent, this util will not be supported anymore and users need to port over its body under their app code.
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) Throws an exception 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) Take an unsignalled exception that is masking another one due to callback failure.
static void	onErrorDropped(Throwable e, Throwable root) Take an unsignalled exception that is masking anowher 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 substraction bound to 0.
static <F> boolean	replace(AtomicReferenceFieldUpdater<F,Subscription> field, F instance, Subscription s) A generic utility to atomically replace a subscription or cancel if marked by a singleton subscription.
static void	reportBadRequest(long n) Throw IllegalArgumentException
static void	reportMoreProduced() Throw IllegalStateException
static void	reportSubscriptionSet() Log reportedSubscriptions
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 a serializing Subscriber.
static <F> boolean	set(AtomicReferenceFieldUpdater<F,Subscription> field, F instance, Subscription s) A generic utility to atomically replace a subscription or cancel if marked by a singleton subscription or concurrently set 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 substraction 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 different null, returning 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

@Deprecated
public static long addAndGet(AtomicLong current,
                                         long toAdd)

Deprecated. This util is neither used by Core nor consistent with the other `addAndGet` methods present which use AtomicLongFieldUpdater. In an effort to keep Operators API consistent, this util will not be supported anymore and users need to port over its body under their app code.

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

Parameters:

current - current atomic to update

toAdd - delta to add

Returns:

Addition result or Long.MAX_VALUE

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

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)

Throws an exception 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 occured

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)

Take an unsignalled exception that is masking anowher one due to callback failure.

Parameters:

e - the exception to handle

root - the optional root cause to suppress

onErrorDropped

public static void onErrorDropped(Throwable e)

Take an unsignalled exception that is masking another one due to callback failure.

Parameters:

e - the exception to handle

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

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. That denotes that an execution was rejected by a Scheduler due to dispose.

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. That denotes that an execution was rejected by a Scheduler due to dispose.

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 substraction bound to 0. 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 sub

Returns:

value after subscription 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 if marked by a singleton subscription.

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 IllegalArgumentException

Parameters:

n - the demand to evaluate

reportMoreProduced

public static void reportMoreProduced()

Throw IllegalStateException

reportSubscriptionSet

public static void reportSubscriptionSet()

Log reportedSubscriptions

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 a serializing Subscriber. 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 wrap

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 if marked by a singleton subscription or concurrently set before.

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.

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 substraction to 0

Parameters:

a - left operand

b - right operand

Returns:

Subscription 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 different null, returning 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