Package 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, Sinks.Many<T>, Sinks.ManyWithUpstream<T>, Scannable

@Deprecated
public final class EmitterProcessor<T>
extends FluxProcessor<T,T>
implements Sinks.ManyWithUpstream<T>

Deprecated.

To be removed in 3.5. Prefer clear cut usage of Sinks through variations of Sinks.many().multicast().onBackpressureBuffer(). If you really need the subscribe-to-upstream functionality of a Processor, switch to Sinks.ManyWithUpstream with Sinks.unsafe() variants of Sinks.unsafe().manyWithUpstream().

This processor was blocking in onNext(Object). This behaviour can be implemented with the Sinks API by calling Sinks.Many.tryEmitNext(Object) and retrying, e.g.:

while (sink.tryEmitNext(v).hasFailed()) {
     LockSupport.parkNanos(10);
 }
 

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.Disposable

Disposable.Composite, Disposable.Swap

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

Scannable.Attr<T>

Field Summary

Fields inherited from interface reactor.core.Scannable

OPERATOR_NAME_UNRELATED_WORDS_PATTERN

Method Summary

Modifier and Type	Method	Description
Flux<T>	asFlux()	Deprecated. Return a Flux view of this sink.
static <E> EmitterProcessor<E>	create()	Deprecated. use Sinks.many().multicast().onBackpressureBuffer() (or the unsafe variant if you're sure about external synchronization).
static <E> EmitterProcessor<E>	create(boolean autoCancel)	Deprecated. use Sinks.many().multicast().onBackpressureBuffer(bufferSize, boolean) using the old default of Queues.SMALL_BUFFER_SIZE for the bufferSize (or the unsafe variant if you're sure about external synchronization).
static <E> EmitterProcessor<E>	create(int bufferSize)	Deprecated. use Sinks.many().multicast().onBackpressureBuffer(bufferSize) (or the unsafe variant if you're sure about external synchronization).
static <E> EmitterProcessor<E>	create(int bufferSize, boolean autoCancel)	Deprecated. use Sinks.many().multicast().onBackpressureBuffer(bufferSize, autoCancel) (or the unsafe variant if you're sure about external synchronization).
Context	currentContext()	Deprecated. Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
int	currentSubscriberCount()	Deprecated. Get how many Subscribers are currently subscribed to the sink.
long	downstreamCount()	Deprecated. Return the number of active Subscriber or -1 if untracked.
default void	emitComplete(Sinks.EmitFailureHandler failureHandler)	A simplified attempt at completing via the Sinks.Many.tryEmitComplete() API, generating an onComplete signal.
default void	emitError(Throwable error, Sinks.EmitFailureHandler failureHandler)	A simplified attempt at failing the sequence via the Sinks.Many.tryEmitError(Throwable) API, generating an onError signal.
default void	emitNext(T value, Sinks.EmitFailureHandler failureHandler)	A simplified attempt at emitting a non-null element via the Sinks.Many.tryEmitNext(Object) API, generating an onNext signal.
final int	getBufferSize()	Deprecated. Return the processor buffer capacity if any or Integer.MAX_VALUE
@Nullable Throwable	getError()	Deprecated. Current error if any, default to null
int	getPending()	Deprecated. Return the number of parked elements in the emitter backlog.
int	getPrefetch()	Deprecated. The prefetch configuration of the Flux
Stream<? extends Scannable>	inners()	Deprecated. Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isCancelled()	Deprecated.
boolean	isDisposed()	Deprecated. Optionally return true when the resource or task is disposed.
protected boolean	isIdentityProcessor()	Deprecated. Return true if FluxProcessor<T, T>
boolean	isTerminated()	Deprecated. Has this upstream finished or "completed" / "failed" ?
void	onComplete()	Deprecated.
void	onError(Throwable throwable)	Deprecated.
void	onNext(T t)	Deprecated.
void	onSubscribe(Subscription s)	Deprecated. Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
@Nullable Object	scanUnsafe(Scannable.Attr key)	Deprecated. This method is used internally by components to define their key-value mappings in a single place.
void	subscribe(CoreSubscriber<? super T> actual)	Deprecated. An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.
Disposable	subscribeTo(Publisher<? extends T> upstream)	Deprecated. Explicitly subscribe this Sinks.Many to an upstream Publisher without exposing it as a Subscriber at all.
Sinks.EmitResult	tryEmitComplete()	Deprecated. Try to terminate the sequence successfully, generating an onComplete signal.
Sinks.EmitResult	tryEmitError(Throwable t)	Deprecated. Try to fail the sequence, generating an onError signal.
Sinks.EmitResult	tryEmitNext(T t)	Deprecated. Try emitting a non-null element, generating an onNext signal.

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, 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, contextCapture, contextWrite, contextWrite, count, create, create, defaultIfEmpty, defer, deferContextual, 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, firstWithSignal, firstWithSignal, firstWithValue, firstWithValue, 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, mapNotNull, materialize, merge, merge, merge, merge, merge, merge, mergeComparing, mergeComparing, mergeComparing, mergeComparingDelayError, mergeComparingWith, mergeDelayError, mergeOrdered, mergeOrdered, mergeOrdered, mergeOrderedWith, mergePriority, mergePriority, mergePriority, mergePriorityDelayError, 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, onErrorComplete, onErrorComplete, onErrorComplete, 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, shareNext, single, single, singleOrEmpty, skip, skip, skip, skipLast, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnFirst, switchOnFirst, switchOnNext, switchOnNext, tag, take, take, take, take, takeLast, takeUntil, takeUntilOther, takeWhile, tap, tap, tap, then, then, thenEmpty, thenMany, timed, timed, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, toString, transform, transformDeferred, transformDeferredContextual, using, using, using, using, usingWhen, usingWhen, window, window, window, window, window, window, window, windowTimeout, windowTimeout, 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, isScanAvailable, name, parents, scan, scanOrDefault, stepName, steps, tags, tagsDeduplicated

Method Details

create

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

Deprecated.

use Sinks.many().multicast().onBackpressureBuffer() (or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.

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

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

Deprecated.

use Sinks.many().multicast().onBackpressureBuffer(bufferSize, boolean) using the old default of Queues.SMALL_BUFFER_SIZE for the bufferSize (or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.

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

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

Deprecated.

use Sinks.many().multicast().onBackpressureBuffer(bufferSize) (or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.

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

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

Deprecated.

use Sinks.many().multicast().onBackpressureBuffer(bufferSize, autoCancel) (or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.

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()

Deprecated.

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()

Deprecated.

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()

subscribeTo

public Disposable subscribeTo(Publisher<? extends T> upstream)

Deprecated.

Description copied from interface: Sinks.ManyWithUpstream

Explicitly subscribe this Sinks.Many to an upstream Publisher without exposing it as a Subscriber at all.

Note that when this is done, one MUST stop using emit/tryEmit APIs, reserving signal creation to be the sole responsibility of the upstream Publisher.

The returned Disposable provides a way of both unsubscribing from the upstream and terminating the sink: currently registered subscribers downstream receive an onError signal with a CancellationException and further attempts at subscribing to the sink will trigger a similar signal immediately (in which case the returned Disposable might be no-op).

Any attempt at subscribing the same Sinks.ManyWithUpstream multiple times throws an IllegalStateException indicating that the subscription must be unique.

Specified by:

subscribeTo in interface Sinks.ManyWithUpstream<T>

subscribe

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

Deprecated.

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)

onComplete

public void onComplete()

Deprecated.

Specified by:

onComplete in interface Subscriber<T>

tryEmitComplete

public Sinks.EmitResult tryEmitComplete()

Deprecated.

Description copied from interface: Sinks.Many

Try to terminate the sequence successfully, generating an onComplete signal. The result of the attempt is represented as an Sinks.EmitResult, which possibly indicates error cases.

See the list of failure Sinks.EmitResult in Sinks.Many.emitComplete(EmitFailureHandler) javadoc for an example of how each of these can be dealt with, to decide if the emit API would be a good enough fit instead.

Specified by:

tryEmitComplete in interface Sinks.Many<T>

Returns:

an Sinks.EmitResult, which should be checked to distinguish different possible failures

See Also:

• Subscriber.onComplete()

onError

public void onError(Throwable throwable)

Deprecated.

Specified by:

onError in interface Subscriber<T>

tryEmitError

public Sinks.EmitResult tryEmitError(Throwable t)

Deprecated.

Description copied from interface: Sinks.Many

Try to fail the sequence, generating an onError signal. The result of the attempt is represented as an Sinks.EmitResult, which possibly indicates error cases.

See the list of failure Sinks.EmitResult in Sinks.Many.emitError(Throwable, EmitFailureHandler) javadoc for an example of how each of these can be dealt with, to decide if the emit API would be a good enough fit instead.

Specified by:

tryEmitError in interface Sinks.Many<T>

Parameters:

t - the exception to signal, not null

Returns:

an Sinks.EmitResult, which should be checked to distinguish different possible failures

See Also:

• Subscriber.onError(Throwable)

onNext

public void onNext(T t)

Deprecated.

Specified by:

onNext in interface Subscriber<T>

tryEmitNext

public Sinks.EmitResult tryEmitNext(T t)

Deprecated.

Description copied from interface: Sinks.Many

Try emitting a non-null element, generating an onNext signal. The result of the attempt is represented as an Sinks.EmitResult, which possibly indicates error cases.

See the list of failure Sinks.EmitResult in Sinks.Many.emitNext(Object, EmitFailureHandler) javadoc for an example of how each of these can be dealt with, to decide if the emit API would be a good enough fit instead.

Might throw an unchecked exception as a last resort (eg. in case of a fatal error downstream which cannot be propagated to any asynchronous handler, a bubbling exception, ...).

Specified by:

tryEmitNext in interface Sinks.Many<T>

Parameters:

t - the value to emit, not null

Returns:

an Sinks.EmitResult, which should be checked to distinguish different possible failures

See Also:

• Subscriber.onNext(Object)

currentSubscriberCount

public int currentSubscriberCount()

Deprecated.

Description copied from interface: Sinks.Many

Get how many Subscribers are currently subscribed to the sink.

This is a best effort peek at the sink state, and a subsequent attempt at emitting to the sink might still return Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER where relevant. (generally in Sinks.Many.tryEmitNext(Object)). Request (and lack thereof) isn't taken into account, all registered subscribers are counted.

Specified by:

currentSubscriberCount in interface Sinks.Many<T>

Returns:

the number of subscribers at the time of invocation

asFlux

public Flux<T> asFlux()

Deprecated.

Description copied from interface: Sinks.Many

Return a Flux view of this sink. Every call returns the same instance.

Specified by:

asFlux in interface Sinks.Many<T>

Returns:

the Flux view associated to this Sinks.Many

isIdentityProcessor

protected boolean isIdentityProcessor()

Deprecated.

Description copied from class: FluxProcessor

Return true if FluxProcessor<T, T>

Overrides:

isIdentityProcessor in class FluxProcessor<T,T>

Returns:

true if FluxProcessor<T, T>

getPending

public int getPending()

Deprecated.

Return the number of parked elements in the emitter backlog.

Returns:

the number of parked elements in the emitter backlog.

isDisposed

public boolean isDisposed()

Deprecated.

Description copied from interface: Disposable

Optionally return true when the resource or task is disposed.

Implementations are not required to track disposition and as such may never return true even when disposed. However, they MUST only return true when there's a guarantee the resource or task is disposed.

Specified by:

isDisposed in interface Disposable

Returns:

true when there's a guarantee the resource or task is disposed.

onSubscribe

public void onSubscribe(Subscription s)

Deprecated.

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.

Specified by:

onSubscribe in interface CoreSubscriber<T>

Specified by:

onSubscribe in interface Subscriber<T>

getError

public @Nullable Throwable getError()

Deprecated.

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()

Deprecated.

Returns:

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

getBufferSize

public final int getBufferSize()

Deprecated.

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()

Deprecated.

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()

Deprecated.

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

public @Nullable Object scanUnsafe(Scannable.Attr key)

Deprecated.

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()

Deprecated.

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

emitNext

default void emitNext(T value,
 Sinks.EmitFailureHandler failureHandler)

Description copied from interface: Sinks.Many

A simplified attempt at emitting a non-null element via the Sinks.Many.tryEmitNext(Object) API, generating an onNext signal. If the result of the attempt is not a success, implementations SHOULD retry the Sinks.Many.tryEmitNext(Object) call IF the provided Sinks.EmitFailureHandler returns true. Otherwise, failures are dealt with in a predefined way that might depend on the actual sink implementation (see below for the vanilla reactor-core behavior).

Generally, Sinks.Many.tryEmitNext(Object) is preferable since it allows a custom handling of error cases, although this implies checking the returned Sinks.EmitResult and correctly acting on it. This API is intended as a good default for convenience.

When the Sinks.EmitResult is not a success, vanilla reactor-core operators have the following behavior:

• Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER: no particular handling. should ideally discard the value but at that point there's no Subscriber from which to get a contextual discard handler.
• Sinks.EmitResult.FAIL_OVERFLOW: discard the value (Operators.onDiscard(Object, Context)) then call Sinks.Many.emitError(Throwable, Sinks.EmitFailureHandler) with a Exceptions.failWithOverflow(String) exception.
• Sinks.EmitResult.FAIL_CANCELLED: discard the value (Operators.onDiscard(Object, Context)).
• Sinks.EmitResult.FAIL_TERMINATED: drop the value (Operators.onNextDropped(Object, Context)).
• Sinks.EmitResult.FAIL_NON_SERIALIZED: throw an Sinks.EmissionException mentioning RS spec rule 1.3. Note that Sinks.unsafe() never trigger this result. It would be possible for an Sinks.EmitFailureHandler to busy-loop and optimistically wait for the contention to disappear to avoid this case for safe sinks...

Might throw an unchecked exception as a last resort (eg. in case of a fatal error downstream which cannot be propagated to any asynchronous handler, a bubbling exception, a Sinks.EmitResult.FAIL_NON_SERIALIZED as described above, ...).

Specified by:

emitNext in interface Sinks.Many<T>

Parameters:

value - the value to emit, not null

failureHandler - the failure handler that allows retrying failed Sinks.EmitResult.

See Also:

• Sinks.Many.tryEmitNext(Object)
• Subscriber.onNext(Object)

emitComplete

default void emitComplete(Sinks.EmitFailureHandler failureHandler)

Description copied from interface: Sinks.Many

A simplified attempt at completing via the Sinks.Many.tryEmitComplete() API, generating an onComplete signal. If the result of the attempt is not a success, implementations SHOULD retry the Sinks.Many.tryEmitComplete() call IF the provided Sinks.EmitFailureHandler returns true. Otherwise, failures are dealt with in a predefined way that might depend on the actual sink implementation (see below for the vanilla reactor-core behavior).

Generally, Sinks.Many.tryEmitComplete() is preferable since it allows a custom handling of error cases, although this implies checking the returned Sinks.EmitResult and correctly acting on it. This API is intended as a good default for convenience.

When the Sinks.EmitResult is not a success, vanilla reactor-core operators have the following behavior:

• Sinks.EmitResult.FAIL_OVERFLOW: irrelevant as onComplete is not driven by backpressure.
• Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER: the completion can be ignored since nobody is listening. Note that most vanilla reactor sinks never trigger this result for onComplete, replaying the terminal signal to later subscribers instead (to the exception of Sinks.UnicastSpec.onBackpressureError()).
• Sinks.EmitResult.FAIL_CANCELLED: the completion can be ignored since nobody is interested.
• Sinks.EmitResult.FAIL_TERMINATED: the extra completion is basically ignored since there was a previous termination signal, but there is nothing interesting to log.
• Sinks.EmitResult.FAIL_NON_SERIALIZED: throw an Sinks.EmissionException mentioning RS spec rule 1.3. Note that Sinks.unsafe() never trigger this result. It would be possible for an Sinks.EmitFailureHandler to busy-loop and optimistically wait for the contention to disappear to avoid this case in safe sinks...

Might throw an unchecked exception as a last resort (eg. in case of a fatal error downstream which cannot be propagated to any asynchronous handler, a bubbling exception, a Sinks.EmitResult.FAIL_NON_SERIALIZED as described above, ...).

Specified by:

emitComplete in interface Sinks.Many<T>

Parameters:

failureHandler - the failure handler that allows retrying failed Sinks.EmitResult.

See Also:

• Sinks.Many.tryEmitComplete()
• Subscriber.onComplete()

emitError

default void emitError(Throwable error,
 Sinks.EmitFailureHandler failureHandler)

Description copied from interface: Sinks.Many

A simplified attempt at failing the sequence via the Sinks.Many.tryEmitError(Throwable) API, generating an onError signal. If the result of the attempt is not a success, implementations SHOULD retry the Sinks.Many.tryEmitError(Throwable) call IF the provided Sinks.EmitFailureHandler returns true. Otherwise, failures are dealt with in a predefined way that might depend on the actual sink implementation (see below for the vanilla reactor-core behavior).

Generally, Sinks.Many.tryEmitError(Throwable) is preferable since it allows a custom handling of error cases, although this implies checking the returned Sinks.EmitResult and correctly acting on it. This API is intended as a good default for convenience.

When the Sinks.EmitResult is not a success, vanilla reactor-core operators have the following behavior:

• Sinks.EmitResult.FAIL_OVERFLOW: irrelevant as onError is not driven by backpressure.
• Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER: the error is ignored since nobody is listening. Note that most vanilla reactor sinks never trigger this result for onError, replaying the terminal signal to later subscribers instead (to the exception of Sinks.UnicastSpec.onBackpressureError()).
• Sinks.EmitResult.FAIL_CANCELLED: the error can be ignored since nobody is interested.
• Sinks.EmitResult.FAIL_TERMINATED: the error unexpectedly follows another terminal signal, so it is dropped via Operators.onErrorDropped(Throwable, Context).
• Sinks.EmitResult.FAIL_NON_SERIALIZED: throw an Sinks.EmissionException mentioning RS spec rule 1.3. Note that Sinks.unsafe() never trigger this result. It would be possible for an Sinks.EmitFailureHandler to busy-loop and optimistically wait for the contention to disappear to avoid this case in safe sinks...

Might throw an unchecked exception as a last resort (eg. in case of a fatal error downstream which cannot be propagated to any asynchronous handler, a bubbling exception, a Sinks.EmitResult.FAIL_NON_SERIALIZED as described above, ...).

Specified by:

emitError in interface Sinks.Many<T>

Parameters:

error - the exception to signal, not null

failureHandler - the failure handler that allows retrying failed Sinks.EmitResult.

See Also:

• Sinks.Many.tryEmitError(Throwable)
• Subscriber.onError(Throwable)