Class MulticastProcessor<T>
- Type Parameters:
T
- the input and output value type
- All Implemented Interfaces:
FlowableSubscriber<T>
,org.reactivestreams.Processor<T,
,T> org.reactivestreams.Publisher<T>
,org.reactivestreams.Subscriber<T>
FlowableProcessor
implementation that coordinates downstream requests through
a front-buffer and stable-prefetching, optionally canceling the upstream if all
subscribers have cancelled.
This processor does not have a public constructor by design; a new empty instance of this
MulticastProcessor
can be created via the following create
methods that
allow configuring it:
create()
: create an emptyMulticastProcessor
withFlowable.bufferSize()
prefetch amount and no reference counting behavior.create(int)
: create an emptyMulticastProcessor
with the given prefetch amount and no reference counting behavior.create(boolean)
: create an emptyMulticastProcessor
withFlowable.bufferSize()
prefetch amount and an optional reference counting behavior.create(int, boolean)
: create an emptyMulticastProcessor
with the given prefetch amount and an optional reference counting behavior.
When the reference counting behavior is enabled, the MulticastProcessor
cancels its
upstream when all Subscriber
s have cancelled. Late Subscriber
s will then be
immediately completed.
Because MulticastProcessor
implements the Subscriber
interface, calling
onSubscribe
is mandatory (Rule 2.12).
If MulticastProcessor
should run standalone, i.e., without subscribing the MulticastProcessor
to another Publisher
,
use start()
or startUnbounded()
methods to initialize the internal buffer.
Failing to do so will lead to a NullPointerException
at runtime.
Use offer(Object)
to try and offer/emit items but don't fail if the
internal buffer is full.
A MulticastProcessor
is a Processor
type in the Reactive Streams specification,
null
s are not allowed (Rule 2.13) as
parameters to onSubscribe(Subscription)
, offer(Object)
, onNext(Object)
and onError(Throwable)
.
Such calls will result in a NullPointerException
being thrown and the processor's state is not changed.
Since a MulticastProcessor
is a Flowable
, it supports backpressure.
The backpressure from the currently subscribed Subscriber
s are coordinated by emitting upstream
items only if all of those Subscriber
s have requested at least one item. This behavior
is also called lockstep-mode because even if some Subscriber
s can take any number
of items, other Subscriber
s requesting less or infrequently will slow down the overall
throughput of the flow.
Calling onNext(Object)
, offer(Object)
, onError(Throwable)
and onComplete()
is required to be serialized (called from the same thread or called non-overlappingly from different threads
through external means of serialization). The FlowableProcessor.toSerialized()
method available to all FlowableProcessor
s
provides such serialization and also protects against reentrance (i.e., when a downstream Subscriber
consuming this processor also wants to call onNext(Object)
on this processor recursively).
This MulticastProcessor
supports the standard state-peeking methods hasComplete()
, hasThrowable()
,
getThrowable()
and hasSubscribers()
. This processor doesn't allow peeking into its buffer.
When this MulticastProcessor
is terminated via onError(Throwable)
or onComplete()
,
all previously signaled but not yet consumed items will be still available to Subscriber
s and the respective
terminal even is only emitted when all previous items have been successfully delivered to Subscriber
s.
If there are no Subscriber
s, the remaining items will be buffered indefinitely.
The MulticastProcessor
does not support clearing its cached events (to appear empty again).
- Backpressure:
- The backpressure from the currently subscribed
Subscriber
s are coordinated by emitting upstream items only if all of thoseSubscriber
s have requested at least one item. This behavior is also called lockstep-mode because even if someSubscriber
s can take any number of items, otherSubscriber
s requesting less or infrequently will slow down the overall throughput of the flow. - Scheduler:
MulticastProcessor
does not operate by default on a particularScheduler
and theSubscriber
s get notified on an arbitrary thread in a serialized fashion.
Example:
MulticastProcessor<Integer> mp = Flowable.range(1, 10)
.subscribeWith(MulticastProcessor.create());
mp.test().assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// --------------------
MulticastProcessor<Integer> mp2 = MulticastProcessor.create(4);
mp2.start();
assertTrue(mp2.offer(1));
assertTrue(mp2.offer(2));
assertTrue(mp2.offer(3));
assertTrue(mp2.offer(4));
assertFalse(mp2.offer(5));
mp2.onComplete();
mp2.test().assertResult(1, 2, 3, 4);
History: 2.1.14 - experimental
- Since:
- 2.2
-
Nested Class Summary
Nested ClassesModifier and TypeClassDescription(package private) static final class
-
Field Summary
FieldsModifier and TypeFieldDescription(package private) final int
(package private) int
(package private) boolean
(package private) static final MulticastProcessor.MulticastSubscription[]
(package private) Throwable
(package private) int
(package private) final int
(package private) SimpleQueue
<T> (package private) final boolean
(package private) final AtomicReference
<MulticastProcessor.MulticastSubscription<T>[]> (package private) static final MulticastProcessor.MulticastSubscription[]
(package private) final AtomicReference
<org.reactivestreams.Subscription> (package private) final AtomicInteger
-
Constructor Summary
ConstructorsConstructorDescriptionMulticastProcessor
(int bufferSize, boolean refCount) Constructs a fresh instance with the given prefetch amount and the optional refCount-behavior. -
Method Summary
Modifier and TypeMethodDescription(package private) boolean
static <T> @NonNull MulticastProcessor
<T> create()
Constructs a fresh instance with the default Flowable.bufferSize() prefetch amount and no refCount-behavior.static <T> @NonNull MulticastProcessor
<T> create
(boolean refCount) Constructs a fresh instance with the default Flowable.bufferSize() prefetch amount and the optional refCount-behavior.static <T> @NonNull MulticastProcessor
<T> create
(int bufferSize) Constructs a fresh instance with the given prefetch amount and no refCount behavior.static <T> @NonNull MulticastProcessor
<T> create
(int bufferSize, boolean refCount) Constructs a fresh instance with the given prefetch amount and the optional refCount-behavior.(package private) void
drain()
Returns the error that caused the FlowableProcessor to terminate or null if the FlowableProcessor hasn't terminated yet.boolean
Returns true if the FlowableProcessor has reached a terminal state through a complete event.boolean
Returns true if the FlowableProcessor has subscribers.boolean
Returns true if the FlowableProcessor has reached a terminal state through an error event.boolean
Tries to offer an item into the internal queue and returns false if the queue is full.void
void
void
void
onSubscribe
(@NonNull org.reactivestreams.Subscription s) Implementors of this method should make sure everything that needs to be visible inSubscriber.onNext(Object)
is established before callingSubscription.request(long)
.(package private) void
void
start()
Initializes this Processor by setting an upstream Subscription that ignores request amounts, uses a fixed buffer and allows using the onXXX and offer methods afterwards.void
Initializes this Processor by setting an upstream Subscription that ignores request amounts, uses an unbounded buffer and allows using the onXXX and offer methods afterwards.protected void
subscribeActual
(@NonNull org.reactivestreams.Subscriber<? super @NonNull T> s) Operator implementations (both source and intermediate) should implement this method that performs the necessary business logic and handles the incomingSubscriber
s.Methods inherited from class io.reactivex.rxjava3.processors.FlowableProcessor
toSerialized
Methods inherited from class io.reactivex.rxjava3.core.Flowable
all, amb, ambArray, ambWith, any, blockingFirst, blockingFirst, blockingForEach, blockingForEach, blockingIterable, blockingIterable, blockingLast, blockingLast, blockingLatest, blockingMostRecent, blockingNext, blockingSingle, blockingSingle, blockingStream, blockingStream, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferSize, cache, cacheWithInitialCapacity, cast, collect, collect, collectInto, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatestArray, combineLatestArray, combineLatestArrayDelayError, combineLatestArrayDelayError, combineLatestDelayError, combineLatestDelayError, compose, concat, concat, concat, concat, concat, concat, concatArray, concatArrayDelayError, concatArrayEager, concatArrayEager, concatArrayEagerDelayError, concatArrayEagerDelayError, concatDelayError, concatDelayError, concatDelayError, concatEager, concatEager, concatEager, concatEager, concatEagerDelayError, concatEagerDelayError, concatEagerDelayError, concatEagerDelayError, concatMap, concatMap, concatMap, concatMapCompletable, concatMapCompletable, concatMapCompletableDelayError, concatMapCompletableDelayError, concatMapCompletableDelayError, concatMapDelayError, concatMapDelayError, concatMapDelayError, concatMapEager, concatMapEager, concatMapEagerDelayError, concatMapEagerDelayError, concatMapIterable, concatMapIterable, concatMapMaybe, concatMapMaybe, concatMapMaybeDelayError, concatMapMaybeDelayError, concatMapMaybeDelayError, concatMapSingle, concatMapSingle, concatMapSingleDelayError, concatMapSingleDelayError, concatMapSingleDelayError, concatMapStream, concatMapStream, concatWith, concatWith, concatWith, concatWith, contains, count, create, debounce, debounce, debounce, debounce, defaultIfEmpty, defer, delay, delay, delay, delay, delay, delay, delaySubscription, delaySubscription, delaySubscription, dematerialize, distinct, distinct, distinct, distinctUntilChanged, distinctUntilChanged, distinctUntilChanged, doAfterNext, doAfterTerminate, doFinally, doOnCancel, doOnComplete, doOnEach, doOnEach, doOnError, doOnLifecycle, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elementAt, elementAt, elementAtOrError, empty, error, error, filter, first, firstElement, firstOrError, firstOrErrorStage, firstStage, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMapCompletable, flatMapCompletable, flatMapIterable, flatMapIterable, flatMapIterable, flatMapIterable, flatMapMaybe, flatMapMaybe, flatMapSingle, flatMapSingle, flatMapStream, flatMapStream, forEach, forEachWhile, forEachWhile, forEachWhile, fromAction, fromArray, fromCallable, fromCompletable, fromCompletionStage, fromFuture, fromFuture, fromIterable, fromMaybe, fromObservable, fromOptional, fromPublisher, fromRunnable, fromSingle, fromStream, fromSupplier, generate, generate, generate, generate, generate, groupBy, groupBy, groupBy, groupBy, groupBy, groupBy, groupJoin, hide, ignoreElements, interval, interval, interval, interval, intervalRange, intervalRange, isEmpty, join, just, just, just, just, just, just, just, just, just, just, last, lastElement, lastOrError, lastOrErrorStage, lastStage, lift, map, mapOptional, materialize, merge, merge, merge, merge, merge, merge, merge, merge, mergeArray, mergeArray, mergeArrayDelayError, mergeArrayDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeWith, mergeWith, mergeWith, mergeWith, never, observeOn, observeOn, observeOn, ofType, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureLatest, onBackpressureLatest, onBackpressureReduce, onBackpressureReduce, onErrorComplete, onErrorComplete, onErrorResumeNext, onErrorResumeWith, onErrorReturn, onErrorReturnItem, onTerminateDetach, parallel, parallel, parallel, publish, publish, publish, publish, range, rangeLong, rebatchRequests, reduce, reduce, reduceWith, repeat, repeat, repeatUntil, repeatWhen, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, retry, retry, retry, retry, retry, retryUntil, retryWhen, safeSubscribe, sample, sample, sample, sample, sample, sample, sample, scan, scan, scanWith, sequenceEqual, sequenceEqual, sequenceEqual, sequenceEqual, serialize, share, single, singleElement, singleOrError, singleOrErrorStage, singleStage, skip, skip, skip, skipLast, skipLast, skipLast, skipLast, skipLast, skipLast, skipUntil, skipWhile, sorted, sorted, startWith, startWith, startWith, startWith, startWithArray, startWithItem, startWithIterable, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchMapCompletable, switchMapCompletableDelayError, switchMapDelayError, switchMapDelayError, switchMapMaybe, switchMapMaybeDelayError, switchMapSingle, switchMapSingleDelayError, switchOnNext, switchOnNext, switchOnNextDelayError, switchOnNextDelayError, take, take, take, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeUntil, takeUntil, takeWhile, test, test, test, throttleFirst, throttleFirst, throttleFirst, throttleLast, throttleLast, throttleLast, throttleLatest, throttleLatest, throttleLatest, throttleLatest, throttleLatest, throttleWithTimeout, throttleWithTimeout, throttleWithTimeout, timeInterval, timeInterval, timeInterval, timeInterval, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timer, timer, timestamp, timestamp, timestamp, timestamp, to, toFuture, toList, toList, toList, toMap, toMap, toMap, toMultimap, toMultimap, toMultimap, toMultimap, toObservable, toSortedList, toSortedList, toSortedList, toSortedList, unsafeCreate, unsubscribeOn, using, using, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zipArray, zipWith, zipWith, zipWith, zipWith
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
Methods inherited from interface org.reactivestreams.Publisher
subscribe
-
Field Details
-
wip
-
upstream
-
subscribers
-
bufferSize
final int bufferSize -
limit
final int limit -
refcount
final boolean refcount -
queue
-
done
volatile boolean done -
error
-
consumed
int consumed -
fusionMode
int fusionMode -
EMPTY
-
TERMINATED
-
-
Constructor Details
-
MulticastProcessor
MulticastProcessor(int bufferSize, boolean refCount) Constructs a fresh instance with the given prefetch amount and the optional refCount-behavior.- Parameters:
bufferSize
- the prefetch amountrefCount
- if true and if all Subscribers have canceled, the upstream is cancelled
-
-
Method Details
-
create
Constructs a fresh instance with the default Flowable.bufferSize() prefetch amount and no refCount-behavior.- Type Parameters:
T
- the input and output value type- Returns:
- the new MulticastProcessor instance
-
create
@CheckReturnValue @NonNull public static <T> @NonNull MulticastProcessor<T> create(boolean refCount) Constructs a fresh instance with the default Flowable.bufferSize() prefetch amount and the optional refCount-behavior.- Type Parameters:
T
- the input and output value type- Parameters:
refCount
- if true and if all Subscribers have canceled, the upstream is cancelled- Returns:
- the new MulticastProcessor instance
-
create
Constructs a fresh instance with the given prefetch amount and no refCount behavior.- Type Parameters:
T
- the input and output value type- Parameters:
bufferSize
- the prefetch amount- Returns:
- the new MulticastProcessor instance
- Throws:
IllegalArgumentException
- ifbufferSize
is non-positive
-
create
@CheckReturnValue @NonNull public static <T> @NonNull MulticastProcessor<T> create(int bufferSize, boolean refCount) Constructs a fresh instance with the given prefetch amount and the optional refCount-behavior.- Type Parameters:
T
- the input and output value type- Parameters:
bufferSize
- the prefetch amountrefCount
- if true and if all Subscribers have canceled, the upstream is cancelled- Returns:
- the new MulticastProcessor instance
- Throws:
IllegalArgumentException
- ifbufferSize
is non-positive
-
start
public void start()Initializes this Processor by setting an upstream Subscription that ignores request amounts, uses a fixed buffer and allows using the onXXX and offer methods afterwards. -
startUnbounded
public void startUnbounded()Initializes this Processor by setting an upstream Subscription that ignores request amounts, uses an unbounded buffer and allows using the onXXX and offer methods afterwards. -
onSubscribe
Description copied from interface:FlowableSubscriber
Implementors of this method should make sure everything that needs to be visible inSubscriber.onNext(Object)
is established before callingSubscription.request(long)
. In practice this means no initialization should happen after therequest()
call and additional behavior is thread safe in respect toonNext
. -
onNext
-
offer
Tries to offer an item into the internal queue and returns false if the queue is full.- Parameters:
t
- the item to offer, notnull
- Returns:
- true if successful, false if the queue is full
- Throws:
NullPointerException
- ift
isnull
IllegalStateException
- if the processor is in fusion mode
-
onError
-
onComplete
public void onComplete() -
hasSubscribers
Description copied from class:FlowableProcessor
Returns true if the FlowableProcessor has subscribers.The method is thread-safe.
- Specified by:
hasSubscribers
in classFlowableProcessor<T>
- Returns:
- true if the FlowableProcessor has subscribers
-
hasThrowable
Description copied from class:FlowableProcessor
Returns true if the FlowableProcessor has reached a terminal state through an error event.The method is thread-safe.
- Specified by:
hasThrowable
in classFlowableProcessor<T>
- Returns:
- true if the FlowableProcessor has reached a terminal state through an error event
- See Also:
-
hasComplete
Description copied from class:FlowableProcessor
Returns true if the FlowableProcessor has reached a terminal state through a complete event.The method is thread-safe.
- Specified by:
hasComplete
in classFlowableProcessor<T>
- Returns:
- true if the FlowableProcessor has reached a terminal state through a complete event
- See Also:
-
getThrowable
Description copied from class:FlowableProcessor
Returns the error that caused the FlowableProcessor to terminate or null if the FlowableProcessor hasn't terminated yet.The method is thread-safe.
- Specified by:
getThrowable
in classFlowableProcessor<T>
- Returns:
- the error that caused the FlowableProcessor to terminate or null if the FlowableProcessor hasn't terminated yet
-
subscribeActual
protected void subscribeActual(@NonNull @NonNull org.reactivestreams.Subscriber<? super @NonNull T> s) Description copied from class:Flowable
Operator implementations (both source and intermediate) should implement this method that performs the necessary business logic and handles the incomingSubscriber
s.There is no need to call any of the plugin hooks on the current
Flowable
instance or theSubscriber
; all hooks and basic safeguards have been applied byFlowable.subscribe(Subscriber)
before this method gets called.- Specified by:
subscribeActual
in classFlowable<T>
- Parameters:
s
- the incomingSubscriber
, nevernull
-
add
-
remove
-
drain
void drain()
-