reactor.core.publisher

Class DirectProcessor<T>

java.lang.Object

reactor.core.publisher.Flux<OUT>

reactor.core.publisher.FluxProcessor<T,T>

reactor.core.publisher.DirectProcessor<T>

Type Parameters:

T - the input and output value type

All Implemented Interfaces:

Processor<T,T>, Publisher<T>, Subscriber<T>, CorePublisher<T>, CoreSubscriber<T>, Disposable, Scannable

public final class DirectProcessor<T>
extends FluxProcessor<T,T>

Dispatches onNext, onError and onComplete signals to zero-to-many Subscribers. Please note, that along with multiple consumers, current implementation of DirectProcessor supports multiple producers. However, all producers must produce messages on the same Thread, otherwise Reactive Streams Spec contract is violated.

Note: DirectProcessor does not coordinate backpressure between its Subscribers and the upstream, but consumes its upstream in an unbounded manner. In the case where a downstream Subscriber is not ready to receive items (hasn't requested yet or enough), it will be terminated with an IllegalStateException. Hence in terms of interaction model, DirectProcessor only supports PUSH from the source through the processor to the Subscribers.

Note: If there are no Subscribers, upstream items are dropped and only the terminal events are retained. A terminated DirectProcessor will emit the terminal signal to late subscribers.

Note: The implementation ignores Subscriptions set via onSubscribe.

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

Method Summary

Modifier and Type	Method and Description
static <E> DirectProcessor<E>	create() Create a new DirectProcessor
long	downstreamCount() Return the number of active Subscriber or -1 if untracked.
Throwable	getError() Current error if any, default to null
int	getPrefetch() The prefetch configuration of the Flux
boolean	hasDownstreams() Return true if any Subscriber is actively subscribed
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
void	onComplete()
void	onError(Throwable t)
void	onNext(T t)
void	onSubscribe(Subscription s) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
void	subscribe(CoreSubscriber<? super T> actual) An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.

Methods inherited from class reactor.core.publisher.FluxProcessor

dispose, getBufferSize, hasCompleted, hasError, isSerialized, scanUnsafe, serialize, serializeAlways, sink, sink, switchOnNext, wrap

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, 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, retry, retry, retryBackoff, retryBackoff, retryBackoff, retryBackoff, retryBackoff, 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, subscribeOn, subscribeOn, subscriberContext, subscriberContext, subscribeWith, switchIfEmpty, switchMap, switchMap, 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

currentContext

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

Method Detail

create

public static <E> DirectProcessor<E> create()

Create a new DirectProcessor

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

downstreamCount

public long downstreamCount()

Description copied from class: FluxProcessor

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

Overrides:

downstreamCount in class FluxProcessor<T,T>

Returns:

the number of active Subscriber or -1 if untracked

getError

@Nullable
public Throwable getError()

Description copied from class: FluxProcessor

Current error if any, default to null

Overrides:

getError in class FluxProcessor<T,T>

Returns:

Current error if any, default to null

getPrefetch

public int getPrefetch()

Description copied from class: Flux

The prefetch configuration of the Flux

Overrides:

getPrefetch in class Flux<T>

Returns:

the prefetch configuration of the Flux, -1 if unspecified

hasDownstreams

public boolean hasDownstreams()

Description copied from class: FluxProcessor

Return true if any Subscriber is actively subscribed

Overrides:

hasDownstreams in class FluxProcessor<T,T>

Returns:

true if any Subscriber is actively subscribed

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

Overrides:

inners in class FluxProcessor<T,T>

Returns:

a Stream of referenced inners (flatmap, multicast etc)

isTerminated

public boolean isTerminated()

Description copied from class: FluxProcessor

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

Overrides:

isTerminated in class FluxProcessor<T,T>

Returns:

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

onComplete

public void onComplete()

onError

public void onError(Throwable t)

onNext

public void onNext(T t)

onSubscribe

public void onSubscribe(Subscription s)

Description copied from interface: CoreSubscriber

Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long). Should further onNext related state modification occur, thread-safety will be required.

Note that an invalid request <= 0 will not produce an onError and will simply be ignored or reported through a debug-enabled Logger.

subscribe

public void subscribe(CoreSubscriber<? super T> actual)

Description copied from class: Flux

An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.

In addition to behave as expected by Publisher.subscribe(Subscriber) in a controlled manner, it supports direct subscribe-time Context passing.

Specified by:

subscribe in interface CorePublisher<T>

Specified by:

subscribe in class Flux<T>

Parameters:

actual - the Subscriber interested into the published sequence

See Also:

Flux.subscribe(Subscriber)