rxjava 1.3.8 · rx.internal.operators.BackpressureUtils

Modifier and Type	Field and Description
pack-priv static final long	COMPLETED_MASK Masks the most significant bit, i.e., 0x8000_0000_0000_0000L.
pack-priv static final long	REQUESTED_MASK Masks the request amount bits, i.e., 0x7FFF_FFFF_FFFF_FFFF.

Access	Constructor and Description
private	BackpressureUtils() Utility class, no instances.

Method Summary

Modifier and Type	Method and Description
public static long	Returns: the capped sum of a and b addCap(long the first value a, long the second value b) Adds two positive longs and caps the result at Long.MAX_VALUE.
public static long	Returns: requested value just prior to successful addition getAndAddRequest(AtomicLong atomic long that should be updated requested, long the number of requests to add to the requested count, positive (not validated) n) Adds `n` (not validated) to `requested` and returns the value prior to addition once the addition is successful (uses CAS semantics).
public static long	Returns: the capped product of a and b multiplyCap(long the first value a, long the second value b) Multiplies two positive longs and caps the result at Long.MAX_VALUE.
public static < the value type to emit T> void	postCompleteDone(AtomicLong the holder of current requested amount requested, Queue<T> the queue holding values to be emitted after completion queue, Subscriber<? super T> the subscriber to receive the values actual) Signals the completion of the main sequence and switches to post-completion replay mode.
public static < the value type in the queue T, the value type to emit R> void	postCompleteDone(AtomicLong the holder of current requested amount requested, Queue<T> the queue holding values to be emitted after completion queue, Subscriber<? super R> the subscriber to receive the values actual, Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted exitTransform) Signals the completion of the main sequence and switches to post-completion replay mode and allows exit transformation on the queued values.
pack-priv static < the value type in the queue T, the value type to emit R> void	postCompleteDrain(AtomicLong the holder of current requested amount requested, Queue<T> the queue holding values to be emitted after completion queue, Subscriber<? super R> the subscriber to receive the values subscriber, Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted exitTransform) Drains the queue based on the outstanding requests in post-completed mode (only!) and allows exit transformation on the queued values.
public static < the value type to emit T> boolean	Returns: true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining. postCompleteRequest(AtomicLong the holder of current requested amount requested, long the value requested; n, Queue<T> the queue holding values to be emitted after completion queue, Subscriber<? super T> the subscriber to receive the values actual) Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests.
public static < the value type in the queue T, the value type to emit R> boolean	Returns: true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining. postCompleteRequest(AtomicLong the holder of current requested amount requested, long the value requested; n, Queue<T> the queue holding values to be emitted after completion queue, Subscriber<? super R> the subscriber to receive the values actual, Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted exitTransform) Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests and allows exit transformation on the queued values.
public static long	Returns: the new requested amount produced(AtomicLong the requested amount holder requested, long the value to subtract from the requested amount, has to be positive (not verified) n) Atomically subtracts a value from the requested amount unless it's at Long.MAX_VALUE.
public static boolean	Returns: true if n is positive validate(long the requested amount n) Validates the requested amount and returns true if it is positive.

Inherited from java.lang.Object:: clone equals finalize getClass hashCode notify notifyAll toString wait wait wait

COMPLETED_MASK	back to summary
pack-priv static final long COMPLETED_MASK Masks the most significant bit, i.e., 0x8000_0000_0000_0000L.

REQUESTED_MASK	back to summary
pack-priv static final long REQUESTED_MASK Masks the request amount bits, i.e., 0x7FFF_FFFF_FFFF_FFFF.

BackpressureUtils	back to summary
private BackpressureUtils() Utility class, no instances.

Method Detail

addCap	back to summary
public static long addCap(long a, long b) Adds two positive longs and caps the result at Long.MAX_VALUE. Parameters a:long the first value b:long the second value Returns:long the capped sum of a and b

addCap

back to summary

public static long addCap(long a, long b)

Adds two positive longs and caps the result at Long.MAX_VALUE.

Parameters

a:long: the first value
b:long: the second value

Returns:long

the capped sum of a and b

getAndAddRequest back to summary

getAndAddRequest	back to summary
public static long getAndAddRequest(AtomicLong requested, long n) Adds `n` (not validated) to `requested` and returns the value prior to addition once the addition is successful (uses CAS semantics). If overflows then sets `requested` field to `Long.MAX_VALUE`. Parameters requested:AtomicLong atomic long that should be updated n:long the number of requests to add to the requested count, positive (not validated) Returns:long requested value just prior to successful addition

public static long getAndAddRequest(AtomicLong requested, long n)

Adds n (not validated) to requested and returns the value prior to addition once the addition is successful (uses CAS semantics). If overflows then sets requested field to Long.MAX_VALUE.

Parameters

requested:AtomicLong: atomic long that should be updated
n:long: the number of requests to add to the requested count, positive (not validated)

Returns:long

requested value just prior to successful addition

multiplyCap	back to summary
public static long multiplyCap(long a, long b) Multiplies two positive longs and caps the result at Long.MAX_VALUE. Parameters a:long the first value b:long the second value Returns:long the capped product of a and b

multiplyCap

back to summary

public static long multiplyCap(long a, long b)

Multiplies two positive longs and caps the result at Long.MAX_VALUE.

Parameters

a:long: the first value
b:long: the second value

Returns:long

the capped product of a and b

postCompleteDone	back to summary
public static <T> void postCompleteDone(AtomicLong requested, Queue<T> queue, Subscriber<? super T> actual) Signals the completion of the main sequence and switches to post-completion replay mode. Don't modify the queue after calling this method! Post-completion backpressure handles the case when a source produces values based on requests when it is active but more values are available even after its completion. In this case, the onCompleted() can't just emit the contents of the queue but has to coordinate with the requested amounts. This requires two distinct modes: active and completed. In active mode, requests flow through and the queue is not accessed but in completed mode, requests no-longer reach the upstream but help in draining the queue. The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't allowed. Parameters <T> the value type to emit requested:AtomicLong the holder of current requested amount queue:Queue<T> the queue holding values to be emitted after completion actual:Subscriber<? super T> the subscriber to receive the values

postCompleteDone

back to summary

public static <T> void postCompleteDone(AtomicLong requested, Queue<T> queue, Subscriber<? super T> actual)

Signals the completion of the main sequence and switches to post-completion replay mode.

Don't modify the queue after calling this method!

Post-completion backpressure handles the case when a source produces values based on requests when it is active but more values are available even after its completion. In this case, the onCompleted() can't just emit the contents of the queue but has to coordinate with the requested amounts. This requires two distinct modes: active and completed. In active mode, requests flow through and the queue is not accessed but in completed mode, requests no-longer reach the upstream but help in draining the queue.

The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't allowed.

Parameters

<T>: the value type to emit
requested:AtomicLong: the holder of current requested amount
queue:Queue<T>: the queue holding values to be emitted after completion
actual:Subscriber<? super T>: the subscriber to receive the values

postCompleteDone	back to summary
public static <T, R> void postCompleteDone(AtomicLong requested, Queue<T> queue, Subscriber<? super R> actual, Func1<? super T, ? extends R> exitTransform) Signals the completion of the main sequence and switches to post-completion replay mode and allows exit transformation on the queued values. Don't modify the queue after calling this method! Post-completion backpressure handles the case when a source produces values based on requests when it is active but more values are available even after its completion. In this case, the onCompleted() can't just emit the contents of the queue but has to coordinate with the requested amounts. This requires two distinct modes: active and completed. In active mode, requests flow through and the queue is not accessed but in completed mode, requests no-longer reach the upstream but help in draining the queue. The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't allowed. Parameters <T> the value type in the queue <R> the value type to emit requested:AtomicLong the holder of current requested amount queue:Queue<T> the queue holding values to be emitted after completion actual:Subscriber<? super R> the subscriber to receive the values exitTransform:Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted

postCompleteDone

back to summary

public static <T, R> void postCompleteDone(AtomicLong requested, Queue<T> queue, Subscriber<? super R> actual, Func1<? super T, ? extends R> exitTransform)

Signals the completion of the main sequence and switches to post-completion replay mode and allows exit transformation on the queued values.

Don't modify the queue after calling this method!

The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't allowed.

Parameters

<T>: the value type in the queue
<R>: the value type to emit
requested:AtomicLong: the holder of current requested amount
queue:Queue<T>: the queue holding values to be emitted after completion
actual:Subscriber<? super R>: the subscriber to receive the values
exitTransform:Func1<? super T, ? extends R>: the transformation to apply on the dequeued value to get the value to be emitted

postCompleteDrain	back to summary
pack-priv static <T, R> void postCompleteDrain(AtomicLong requested, Queue<T> queue, Subscriber<? super R> subscriber, Func1<? super T, ? extends R> exitTransform) Drains the queue based on the outstanding requests in post-completed mode (only!) and allows exit transformation on the queued values. Parameters <T> the value type in the queue <R> the value type to emit requested:AtomicLong the holder of current requested amount queue:Queue<T> the queue holding values to be emitted after completion subscriber:Subscriber<? super R> the subscriber to receive the values exitTransform:Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted

postCompleteDrain

back to summary

pack-priv static <T, R> void postCompleteDrain(AtomicLong requested, Queue<T> queue, Subscriber<? super R> subscriber, Func1<? super T, ? extends R> exitTransform)

Drains the queue based on the outstanding requests in post-completed mode (only!) and allows exit transformation on the queued values.

Parameters

<T>: the value type in the queue
<R>: the value type to emit
requested:AtomicLong: the holder of current requested amount
queue:Queue<T>: the queue holding values to be emitted after completion
subscriber:Subscriber<? super R>: the subscriber to receive the values
exitTransform:Func1<? super T, ? extends R>: the transformation to apply on the dequeued value to get the value to be emitted

postCompleteRequest	back to summary
public static <T> boolean postCompleteRequest(AtomicLong requested, long n, Queue<T> queue, Subscriber<? super T> actual) Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests. Post-completion backpressure handles the case when a source produces values based on requests when it is active but more values are available even after its completion. In this case, the onCompleted() can't just emit the contents of the queue but has to coordinate with the requested amounts. This requires two distinct modes: active and completed. In active mode, requests flow through and the queue is not accessed but in completed mode, requests no-longer reach the upstream but help in draining the queue. Parameters <T> the value type to emit requested:AtomicLong the holder of current requested amount n:long the value requested; queue:Queue<T> the queue holding values to be emitted after completion actual:Subscriber<? super T> the subscriber to receive the values Returns:boolean true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining.

postCompleteRequest

back to summary

public static <T> boolean postCompleteRequest(AtomicLong requested, long n, Queue<T> queue, Subscriber<? super T> actual)

Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests.

Parameters

<T>: the value type to emit
requested:AtomicLong: the holder of current requested amount
n:long: the value requested;
queue:Queue<T>: the queue holding values to be emitted after completion
actual:Subscriber<? super T>: the subscriber to receive the values

Returns:boolean

true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining.

postCompleteRequest	back to summary
public static <T, R> boolean postCompleteRequest(AtomicLong requested, long n, Queue<T> queue, Subscriber<? super R> actual, Func1<? super T, ? extends R> exitTransform) Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests and allows exit transformation on the queued values. Post-completion backpressure handles the case when a source produces values based on requests when it is active but more values are available even after its completion. In this case, the onCompleted() can't just emit the contents of the queue but has to coordinate with the requested amounts. This requires two distinct modes: active and completed. In active mode, requests flow through and the queue is not accessed but in completed mode, requests no-longer reach the upstream but help in draining the queue. Parameters <T> the value type in the queue <R> the value type to emit requested:AtomicLong the holder of current requested amount n:long the value requested; queue:Queue<T> the queue holding values to be emitted after completion actual:Subscriber<? super R> the subscriber to receive the values exitTransform:Func1<? super T, ? extends R> the transformation to apply on the dequeued value to get the value to be emitted Returns:boolean true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining.

postCompleteRequest

back to summary

public static <T, R> boolean postCompleteRequest(AtomicLong requested, long n, Queue<T> queue, Subscriber<? super R> actual, Func1<? super T, ? extends R> exitTransform)

Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests and allows exit transformation on the queued values.

Parameters

<T>: the value type in the queue
<R>: the value type to emit
requested:AtomicLong: the holder of current requested amount
n:long: the value requested;
queue:Queue<T>: the queue holding values to be emitted after completion
actual:Subscriber<? super R>: the subscriber to receive the values
exitTransform:Func1<? super T, ? extends R>: the transformation to apply on the dequeued value to get the value to be emitted

Returns:boolean

true if in the active mode and the request amount of n can be relayed to upstream, false if in the post-completed mode and the queue is draining.

produced	back to summary
public static long produced(AtomicLong requested, long n) Atomically subtracts a value from the requested amount unless it's at Long.MAX_VALUE. Parameters requested:AtomicLong the requested amount holder n:long the value to subtract from the requested amount, has to be positive (not verified) Returns:long the new requested amount Exceptions IllegalStateException: if n is greater than the current requested amount, which indicates a bug in the request accounting logic

produced

back to summary

public static long produced(AtomicLong requested, long n)

Atomically subtracts a value from the requested amount unless it's at Long.MAX_VALUE.

Parameters

requested:AtomicLong: the requested amount holder
n:long: the value to subtract from the requested amount, has to be positive (not verified)

Returns:long

the new requested amount

Exceptions

IllegalStateException:: if n is greater than the current requested amount, which indicates a bug in the request accounting logic

validate	back to summary
public static boolean validate(long n) Validates the requested amount and returns true if it is positive. Parameters n:long the requested amount Returns:boolean true if n is positive Exceptions IllegalArgumentException: if n is negative

validate

back to summary

public static boolean validate(long n)

Validates the requested amount and returns true if it is positive.

Parameters

n:long: the requested amount

Returns:boolean

true if n is positive

Exceptions

IllegalArgumentException:: if n is negative

public final Class BackpressureUtils

Field Summary

Constructor Summary

Method Summary

Field Detail

Constructor Detail

Method Detail