reactor.core.publisher

Class FluxProcessor<IN,OUT>

java.lang.Object

reactor.core.publisher.Flux<OUT>

reactor.core.publisher.FluxProcessor<IN,OUT>

Type Parameters:

IN - the input value type

OUT - the output value type

All Implemented Interfaces:

Processor<IN,OUT>, Publisher<OUT>, Subscriber<IN>, CorePublisher<OUT>, CoreSubscriber<IN>, Disposable, Scannable

Direct Known Subclasses:

DirectProcessor, EmitterProcessor, ReplayProcessor, UnicastProcessor

public abstract class FluxProcessor<IN,OUT>
extends Flux<OUT>
implements Processor<IN,OUT>, CoreSubscriber<IN>, Scannable, Disposable

A base processor that exposes Flux API for Processor. Implementors include UnicastProcessor, EmitterProcessor, ReplayProcessor.

Author:

Stephane Maldini

Nested Class Summary

Nested classes/interfaces inherited from interface reactor.core.Scannable

Scannable.Attr<T>

Nested classes/interfaces inherited from interface reactor.core.Disposable

Disposable.Composite, Disposable.Swap

Field Summary

Fields inherited from interface reactor.core.Scannable

OPERATOR_NAME_UNRELATED_WORDS_PATTERN

Constructor Summary

Constructors

Constructor and Description
FluxProcessor()

Method Summary

Modifier and Type	Method and Description
Context	currentContext() Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
void	dispose() Cancel or dispose the underlying task or resource.
long	downstreamCount() Return the number of active Subscriber or -1 if untracked.
int	getBufferSize() Return the processor buffer capacity if any or Integer.MAX_VALUE
Throwable	getError() Current error if any, default to null
boolean	hasCompleted() Return true if terminated with onComplete
boolean	hasDownstreams() Return true if any Subscriber is actively subscribed
boolean	hasError() Return true if terminated with onError
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isSerialized() Return true if this FluxProcessor supports multithread producing
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
Object	scanUnsafe(Scannable.Attr key) This method is used internally by components to define their key-value mappings in a single place.
FluxProcessor<IN,OUT>	serialize() Create a FluxProcessor that safely gates multi-threaded producer Subscriber.onNext(Object).
protected boolean	serializeAlways() Returns serialization strategy.
FluxSink<IN>	sink() Create a FluxSink that safely gates multi-threaded producer Subscriber.onNext(Object).
FluxSink<IN>	sink(FluxSink.OverflowStrategy strategy) Create a FluxSink that safely gates multi-threaded producer Subscriber.onNext(Object).
static <T> FluxProcessor<Publisher<? extends T>,T>	switchOnNext() Build a FluxProcessor whose data are emitted by the most recent emitted Publisher.
static <IN,OUT> FluxProcessor<IN,OUT>	wrap(Subscriber<IN> upstream, Publisher<OUT> downstream) Transform a receiving Subscriber and a producing Publisher in a logical FluxProcessor.

Methods inherited from class reactor.core.publisher.Flux

all, any, as, blockFirst, blockFirst, blockLast, blockLast, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferTimeout, bufferTimeout, bufferTimeout, bufferTimeout, bufferUntil, bufferUntil, bufferUntilChanged, bufferUntilChanged, bufferUntilChanged, bufferWhen, bufferWhen, bufferWhile, cache, cache, cache, cache, cache, cache, cancelOn, cast, checkpoint, checkpoint, checkpoint, collect, collect, collectList, collectMap, collectMap, collectMap, collectMultimap, collectMultimap, collectMultimap, collectSortedList, collectSortedList, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, concat, concat, concat, concat, concatDelayError, concatDelayError, concatDelayError, concatDelayError, concatMap, concatMap, concatMapDelayError, concatMapDelayError, concatMapDelayError, concatMapIterable, concatMapIterable, concatWith, concatWithValues, count, create, create, defaultIfEmpty, defer, deferWithContext, delayElements, delayElements, delaySequence, delaySequence, delaySubscription, delaySubscription, delaySubscription, delayUntil, dematerialize, distinct, distinct, distinct, distinct, distinctUntilChanged, distinctUntilChanged, distinctUntilChanged, doAfterTerminate, doFinally, doFirst, doOnCancel, doOnComplete, doOnDiscard, doOnEach, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, error, error, error, expand, expand, expandDeep, expandDeep, filter, filterWhen, filterWhen, first, first, flatMap, flatMap, flatMap, flatMap, flatMapDelayError, flatMapIterable, flatMapIterable, flatMapSequential, flatMapSequential, flatMapSequential, flatMapSequentialDelayError, from, fromArray, fromIterable, fromStream, fromStream, generate, generate, generate, getPrefetch, groupBy, groupBy, groupBy, groupBy, groupJoin, handle, hasElement, hasElements, hide, ignoreElements, index, index, interval, interval, interval, interval, join, just, just, last, last, limitRate, limitRate, limitRequest, log, log, log, log, log, log, map, materialize, merge, merge, merge, merge, merge, merge, mergeDelayError, mergeOrdered, mergeOrdered, mergeOrdered, mergeOrderedWith, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequentialDelayError, mergeSequentialDelayError, mergeSequentialDelayError, mergeWith, metrics, name, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorContinue, onErrorContinue, onErrorContinue, onErrorMap, onErrorMap, onErrorMap, onErrorResume, onErrorResume, onErrorResume, onErrorReturn, onErrorReturn, onErrorReturn, onErrorStop, onTerminateDetach, or, parallel, parallel, parallel, publish, publish, publish, publish, publishNext, publishOn, publishOn, publishOn, push, push, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, replay, replay, replay, replay, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, single, single, singleOrEmpty, skip, skip, skip, skipLast, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeOn, subscriberContext, subscriberContext, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnFirst, switchOnFirst, switchOnNext, switchOnNext, tag, take, take, take, takeLast, takeUntil, takeUntilOther, takeWhile, then, then, thenEmpty, thenMany, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, toString, transform, transformDeferred, using, using, usingWhen, usingWhen, window, window, window, window, window, window, window, windowTimeout, windowTimeout, windowUntil, windowUntil, windowUntil, windowUntilChanged, windowUntilChanged, windowUntilChanged, windowWhen, windowWhile, windowWhile, withLatestFrom, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zipWith, zipWith, zipWith, zipWith, zipWithIterable, zipWithIterable

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.reactivestreams.Publisher

subscribe

Methods inherited from interface reactor.core.CoreSubscriber

onSubscribe

Methods inherited from interface org.reactivestreams.Subscriber

onComplete, onError, onNext

Methods inherited from interface reactor.core.Scannable

actuals, from, isScanAvailable, name, parents, scan, scanOrDefault, stepName, steps, tags

Methods inherited from interface reactor.core.Disposable

isDisposed

Constructor Detail

FluxProcessor

public FluxProcessor()

Method Detail

currentContext

public Context currentContext()

Description copied from interface: CoreSubscriber

Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.

Specified by:

currentContext in interface CoreSubscriber<IN>

Returns:

a resolved context or Context.empty()

dispose

public void dispose()

Description copied from interface: Disposable

Cancel or dispose the underlying task or resource.

Implementations are required to make this method idempotent.

Specified by:

dispose in interface Disposable

downstreamCount

public long downstreamCount()

Return the number of active Subscriber or -1 if untracked.

Returns:

the number of active Subscriber or -1 if untracked

getBufferSize

public int getBufferSize()

Return the processor buffer capacity if any or Integer.MAX_VALUE

Returns:

processor buffer capacity if any or Integer.MAX_VALUE

getError

@Nullable
public Throwable getError()

Current error if any, default to null

Returns:

Current error if any, default to null

hasCompleted

public final boolean hasCompleted()

Return true if terminated with onComplete

Returns:

true if terminated with onComplete

hasDownstreams

public boolean hasDownstreams()

Return true if any Subscriber is actively subscribed

Returns:

true if any Subscriber is actively subscribed

hasError

public final boolean hasError()

Return true if terminated with onError

Returns:

true if terminated with onError

inners

public Stream<? extends Scannable> inners()

Description copied from interface: Scannable

Return a Stream of referenced inners (flatmap, multicast etc)

Specified by:

inners in interface Scannable

Returns:

a Stream of referenced inners (flatmap, multicast etc)

isSerialized

public boolean isSerialized()

Return true if this FluxProcessor supports multithread producing

Returns:

true if this FluxProcessor supports multithread producing

isTerminated

public boolean isTerminated()

Has this upstream finished or "completed" / "failed" ?

Returns:

has this upstream finished or "completed" / "failed" ?

scanUnsafe

@Nullable
public Object scanUnsafe(Scannable.Attr key)

Description copied from interface: Scannable

This method is used internally by components to define their key-value mappings in a single place. Although it is ignoring the generic type of the Scannable.Attr key, implementors should take care to return values of the correct type, and return null if no specific value is available.

For public consumption of attributes, prefer using Scannable.scan(Attr), which will return a typed value and fall back to the key's default if the component didn't define any mapping.

Specified by:

scanUnsafe in interface Scannable

Parameters:

key - a Scannable.Attr to resolve for the component.

Returns:

the value associated to the key for that specific component, or null if none.

serialize

public final FluxProcessor<IN,OUT> serialize()

Create a FluxProcessor that safely gates multi-threaded producer Subscriber.onNext(Object).

Returns:

a serializing FluxProcessor

onDiscard Support

The resulting processor discards elements received from the source Publisher (if any) when it cancels subscription to said source.

serializeAlways

protected boolean serializeAlways()

Returns serialization strategy. If true, sink() will always be serialized. Otherwise sink is serialized only if FluxSink.onRequest(java.util.function.LongConsumer) is invoked.

Returns:

true to serialize any sink, false to delay serialization till onRequest

sink

public final FluxSink<IN> sink()

Create a FluxSink that safely gates multi-threaded producer Subscriber.onNext(Object). This processor will be subscribed to that FluxSink, and any previous subscribers will be unsubscribed.

The returned FluxSink will not apply any FluxSink.OverflowStrategy and overflowing FluxSink.next(Object) will behave in two possible ways depending on the Processor:

• an unbounded processor will handle the overflow itself by dropping or buffering
• a bounded processor will block/spin

Returns:

a serializing FluxSink

sink

public final FluxSink<IN> sink(FluxSink.OverflowStrategy strategy)

Create a FluxSink that safely gates multi-threaded producer Subscriber.onNext(Object). This processor will be subscribed to that FluxSink, and any previous subscribers will be unsubscribed.

The returned FluxSink will not apply any FluxSink.OverflowStrategy and overflowing FluxSink.next(Object) will behave in two possible ways depending on the Processor:

• an unbounded processor will handle the overflow itself by dropping or buffering
• a bounded processor will block/spin on IGNORE strategy, or apply the strategy behavior

Parameters:

strategy - the overflow strategy, see FluxSink.OverflowStrategy for the available strategies

Returns:

a serializing FluxSink

switchOnNext

public static <T> FluxProcessor<Publisher<? extends T>,T> switchOnNext()

Build a FluxProcessor whose data are emitted by the most recent emitted Publisher. The Flux will complete once both the publishers source and the last switched to Publisher have completed.

Type Parameters:

T - the produced type

Returns:

a FluxProcessor accepting publishers and producing T

wrap

public static <IN,OUT> FluxProcessor<IN,OUT> wrap(Subscriber<IN> upstream,
                                                  Publisher<OUT> downstream)

Transform a receiving Subscriber and a producing Publisher in a logical FluxProcessor. The link between the passed upstream and returned downstream will not be created automatically, e.g. not subscribed together. A Processor might choose to have orthogonal sequence input and output.

Type Parameters:

IN - the receiving type

OUT - the producing type

Parameters:

upstream - the upstream subscriber

downstream - the downstream publisher

Returns:

a new blackboxed FluxProcessor