reactor.core.publisher

Class EmitterProcessor<T>

java.lang.Object

reactor.core.publisher.Flux<OUT>

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

reactor.core.publisher.EmitterProcessor<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 EmitterProcessor<T>
extends FluxProcessor<T,T>

An implementation of a message-passing Processor implementing publish-subscribe with synchronous (thread-stealing and happen-before interactions) drain loops.

The default create() factories will only produce the new elements observed in the parent sequence after a given Subscriber is subscribed.

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

Method Summary

Modifier and Type	Method and Description
static <E> EmitterProcessor<E>	create() Create a new EmitterProcessor using Queues.SMALL_BUFFER_SIZE backlog size and auto-cancel.
static <E> EmitterProcessor<E>	create(boolean autoCancel) Create a new EmitterProcessor using Queues.SMALL_BUFFER_SIZE backlog size and the provided auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize) Create a new EmitterProcessor using the provided backlog size, with auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize, boolean autoCancel) Create a new EmitterProcessor using the provided backlog size and auto-cancellation.
Context	currentContext() Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
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
int	getPending() Return the number of parked elements in the emitter backlog.
int	getPrefetch() The prefetch configuration of the Flux
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isCancelled()
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).
Object	scanUnsafe(Scannable.Attr key) This method is used internally by components to define their key-value mappings in a single place.
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, hasCompleted, hasDownstreams, hasError, isSerialized, 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, compose, 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, onLastAssembly, 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, 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, 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, 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.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> EmitterProcessor<E> create()

Create a new EmitterProcessor using Queues.SMALL_BUFFER_SIZE backlog size and auto-cancel.

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

create

public static <E> EmitterProcessor<E> create(boolean autoCancel)

Create a new EmitterProcessor using Queues.SMALL_BUFFER_SIZE backlog size and the provided auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

autoCancel - automatically cancel

Returns:

a fresh processor

create

public static <E> EmitterProcessor<E> create(int bufferSize)

Create a new EmitterProcessor using the provided backlog size, with auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

bufferSize - the internal buffer size to hold signals

Returns:

a fresh processor

create

public static <E> EmitterProcessor<E> create(int bufferSize,
                                             boolean autoCancel)

Create a new EmitterProcessor using the provided backlog size and auto-cancellation.

Type Parameters:

E - Type of processed signals

Parameters:

bufferSize - the internal buffer size to hold signals

autoCancel - automatically cancel

Returns:

a fresh processor

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)

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<T>

Overrides:

currentContext in class FluxProcessor<T,T>

Returns:

a resolved context or Context.empty()

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)

getPending

public int getPending()

Return the number of parked elements in the emitter backlog.

Returns:

the number of parked elements in the emitter backlog.

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.

onNext

public void onNext(T t)

onError

public void onError(Throwable t)

onComplete

public void onComplete()

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

isCancelled

public boolean isCancelled()

Returns:

true if all subscribers have actually been cancelled and the processor auto shut down

getBufferSize

public final int getBufferSize()

Description copied from class: FluxProcessor

Return the processor buffer capacity if any or Integer.MAX_VALUE

Overrides:

getBufferSize in class FluxProcessor<T,T>

Returns:

processor buffer capacity if any or Integer.MAX_VALUE

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" ?

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

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

Overrides:

scanUnsafe in class FluxProcessor<T,T>

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.

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