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>, MultiProducer, Receiver, Trackable

public final class EmitterProcessor<T>
extends FluxProcessor<T,T>
implements MultiProducer, Receiver

An implementation of a RingBuffer backed 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

Field Summary

Fields inherited from interface reactor.core.Trackable

UNSPECIFIED

Method Summary

Modifier and Type	Method and Description
EmitterProcessor<T>	connect() Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving onNext, onComplete and onError calls.
static <E> EmitterProcessor<E>	create() Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> EmitterProcessor<E>	create(boolean autoCancel) Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize) Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize, boolean autoCancel) Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize, int concurrency) Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> EmitterProcessor<E>	create(int bufferSize, int concurrency, boolean autoCancel) Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
long	downstreamCount() the number of downstream receivers
Iterator<?>	downstreams() the connected data receivers
long	expectedFromUpstream()
long	getCapacity() Return defined element capacity
Throwable	getError() Current error if any, default to null
long	getPending() Return current used space in buffer
boolean	isCancelled()
boolean	isStarted() Has this upstream started or "onSubscribed" ?
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
long	limit()
void	onComplete()
void	onError(Throwable t)
void	onNext(T t)
void	onSubscribe(Subscription s)
void	subscribe(Subscriber<? super T> s)
String	toString()
Subscription	upstream() Return the direct source of data, Supports reference.

Methods inherited from class reactor.core.publisher.FluxProcessor

connectSink, connectSink, serialize, switchOnNext, wrap

Methods inherited from class reactor.core.publisher.Flux

all, any, as, awaitOnSubscribe, blockFirst, blockFirst, blockFirstMillis, blockLast, blockLast, blockLastMillis, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferTimeout, bufferTimeout, bufferTimeoutMillis, bufferTimeoutMillis, bufferTimeoutMillis, bufferUntil, bufferUntil, bufferWhile, cache, cache, cache, cache, cancelOn, cast, 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, count, create, create, defaultIfEmpty, defer, delay, delayElements, delayElementsMillis, delayElementsMillis, delayMillis, delayMillis, delaySubscription, delaySubscription, delaySubscriptionMillis, delaySubscriptionMillis, dematerialize, distinct, distinct, distinctUntilChanged, distinctUntilChanged, doAfterTerminate, doFinally, doOnCancel, doOnComplete, doOnEach, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, error, error, filter, firstEmitting, firstEmitting, firstEmittingWith, flatMap, flatMap, flatMap, flatMap, flatMap, flatMapIterable, flatMapIterable, flatMapSequential, flatMapSequential, flatMapSequential, flatMapSequential, from, fromArray, fromIterable, fromStream, generate, generate, generate, getPrefetch, groupBy, groupBy, groupBy, groupBy, groupJoin, handle, hasElement, hasElements, hide, ignoreElements, interval, interval, intervalMillis, intervalMillis, intervalMillis, intervalMillis, join, just, just, last, last, limitRate, log, log, log, log, map, mapError, mapError, mapError, materialize, merge, merge, merge, merge, merge, merge, merge, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeWith, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorResumeWith, onErrorResumeWith, onErrorResumeWith, onErrorReturn, onErrorReturn, onErrorReturn, onTerminateDetach, parallel, parallel, parallel, publish, publish, publish, publish, publishNext, publishOn, publishOn, publishOn, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, replay, replay, replayMillis, replayMillis, retry, retry, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleFirstMillis, sampleMillis, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, single, single, singleOrEmpty, skip, skip, skipLast, skipMillis, skipMillis, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnError, switchOnError, switchOnError, switchOnNext, switchOnNext, take, take, takeLast, takeMillis, takeMillis, takeUntil, takeUntilOther, takeWhile, then, then, then, thenEmpty, thenMany, thenMany, timeout, timeout, timeout, timeout, timeout, timeoutMillis, timeoutMillis, timeoutMillis, timeoutMillis, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, transform, using, using, window, window, window, window, window, window, window, window, windowMillis, windowMillis, windowMillis, windowMillis, windowMillis, windowTimeout, windowTimeoutMillis, windowTimeoutMillis, windowUntil, windowUntil, windowUntil, windowWhile, windowWhile, withLatestFrom, 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 reactor.core.MultiProducer

hasDownstreams

Methods inherited from interface reactor.core.Trackable

requestedFromDownstream

Method Detail

create

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

Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy 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 QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and 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 QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and 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,
                                             int concurrency)

Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

bufferSize - the internal buffer size to hold signals

concurrency - the concurrency level of the emission

Returns:

a fresh processor

create

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

Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

bufferSize - the internal buffer size to hold signals

autoCancel - automatically cancel

Returns:

a fresh processor

create

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

Create a new EmitterProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

bufferSize - the internal buffer size to hold signals

concurrency - the concurrency level of the emission

autoCancel - automatically cancel

Returns:

a fresh processor

upstream

public Subscription upstream()

Description copied from interface: Receiver

Return the direct source of data, Supports reference.

Specified by:

upstream in interface Receiver

Returns:

the direct source of data, Supports reference.

subscribe

public void subscribe(Subscriber<? super T> s)

Specified by:

subscribe in interface Publisher<T>

Overrides:

subscribe in class FluxProcessor<T,T>

connect

public EmitterProcessor<T> connect()

Description copied from class: FluxProcessor

Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving onNext, onComplete and onError calls.

Doing so MAY allow direct UNBOUNDED onXXX calls and MAY prevent Publisher to subscribe this subscriber. Note that Processor can extend this behavior to effectively start its subscribers.

Overrides:

connect in class FluxProcessor<T,T>

Returns:

this

getPending

public long getPending()

Description copied from interface: Trackable

Return current used space in buffer

Specified by:

getPending in interface Trackable

Returns:

long capacity

onNext

public void onNext(T t)

Specified by:

onNext in interface Subscriber<T>

onError

public void onError(Throwable t)

Specified by:

onError in interface Subscriber<T>

onComplete

public void onComplete()

Specified by:

onComplete in interface Subscriber<T>

onSubscribe

public void onSubscribe(Subscription s)

Specified by:

onSubscribe in interface Subscriber<T>

getError

public Throwable getError()

Description copied from interface: Trackable

Current error if any, default to null

Specified by:

getError in interface Trackable

Returns:

Current error if any, default to null

isCancelled

public boolean isCancelled()

Specified by:

isCancelled in interface Trackable

Returns:

has the downstream "cancelled" and interrupted its consuming ?

getCapacity

public final long getCapacity()

Description copied from interface: Trackable

Return defined element capacity

Specified by:

getCapacity in interface Trackable

Overrides:

getCapacity in class FluxProcessor<T,T>

Returns:

long capacity

isStarted

public boolean isStarted()

Description copied from interface: Trackable

Has this upstream started or "onSubscribed" ?

Specified by:

isStarted in interface Trackable

Returns:

has this upstream started or "onSubscribed" ?

isTerminated

public boolean isTerminated()

Description copied from interface: Trackable

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

Specified by:

isTerminated in interface Trackable

Returns:

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

limit

public long limit()

Specified by:

limit in interface Trackable

Returns:

a given limit threshold to replenish outstanding upstream request

expectedFromUpstream

public long expectedFromUpstream()

Specified by:

expectedFromUpstream in interface Trackable

Returns:

expected number of events to be produced to this component

downstreams

public Iterator<?> downstreams()

Description copied from interface: MultiProducer

the connected data receivers

Specified by:

downstreams in interface MultiProducer

Returns:

the connected data receivers

downstreamCount

public long downstreamCount()

Description copied from interface: MultiProducer

the number of downstream receivers

Specified by:

downstreamCount in interface MultiProducer

Returns:

the number of downstream receivers

toString

public String toString()

Overrides:

toString in class Flux<T>