T
- the input and output typeSinks
through
variations under Sinks.many().unicast()
.@Deprecated public final class UnicastProcessor<T> extends FluxProcessor<T,T> implements Fuseable.QueueSubscription<T>, Fuseable
Note: UnicastProcessor does not respect the actual subscriber's demand as it is described in Reactive Streams Spec. However, UnicastProcessor embraces configurable Queue internally which allows enabling backpressure support and preventing of consumer's overwhelming. Hence, interaction model between producers and UnicastProcessor will be PUSH only. In opposite, interaction model between UnicastProcessor and consumer will be PUSH-PULL as defined in Reactive Streams Spec. In the case when upstream's signals overflow the bound of internal Queue, UnicastProcessor will fail with signaling onError( reactor.core.Exceptions.OverflowException).
Note: The implementation keeps the order of signals. That means that in case of terminal signal (completion or error signals) it will be postponed until all of the previous signals has been consumed.
Fuseable.ConditionalSubscriber<T>, Fuseable.QueueSubscription<T>, Fuseable.ScalarCallable<T>, Fuseable.SynchronousSubscription<T>
Scannable.Attr<T>
Disposable.Composite, Disposable.Swap
NOT_SUPPORTED_MESSAGE
OPERATOR_NAME_UNRELATED_WORDS_PATTERN
Constructor and Description |
---|
UnicastProcessor(Queue<T> queue)
Deprecated.
|
UnicastProcessor(Queue<T> queue,
Consumer<? super T> onOverflow,
Disposable onTerminate)
Deprecated.
|
UnicastProcessor(Queue<T> queue,
Disposable onTerminate)
Deprecated.
|
Modifier and Type | Method and Description |
---|---|
CoreSubscriber<? super T> |
actual()
Deprecated.
|
Flux<T> |
asFlux()
Deprecated.
Return a
Flux view of this sink. |
void |
cancel()
Deprecated.
|
void |
clear()
Deprecated.
|
static <E> UnicastProcessor<E> |
create()
Deprecated.
use
Sinks.many().unicast().onBackpressureBuffer()
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5. |
static <E> UnicastProcessor<E> |
create(Queue<E> queue)
Deprecated.
use
Sinks.many().unicast().onBackpressureBuffer(queue)
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5. |
static <E> UnicastProcessor<E> |
create(Queue<E> queue,
Consumer<? super E> onOverflow,
Disposable endcallback)
Deprecated.
use
Sinks.many().unicast().onBackpressureBuffer(queue, endCallback)
(or the unsafe variant if you're sure about external synchronization). The onOverflow callback is not
supported anymore. To be removed in 3.5. |
static <E> UnicastProcessor<E> |
create(Queue<E> queue,
Disposable endcallback)
Deprecated.
use
Sinks.many().unicast().onBackpressureBuffer(queue, endCallback)
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5. |
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. |
void |
emitNext(T value,
Sinks.EmitFailureHandler failureHandler)
Deprecated.
A simplified attempt at emitting a non-null element via the
Sinks.Many.tryEmitNext(Object) API, generating an
onNext signal. |
int |
getBufferSize()
Deprecated.
Return the processor buffer capacity if any or
Integer.MAX_VALUE |
Throwable |
getError()
Deprecated.
Current error if any, default to null
|
int |
getPrefetch()
Deprecated.
The prefetch configuration of the
Flux |
boolean |
hasDownstreams()
Deprecated.
Return true if any
Subscriber is actively subscribed |
Stream<Scannable> |
inners()
Deprecated.
Return a
Stream of referenced inners (flatmap, multicast etc) |
boolean |
isDisposed()
Deprecated.
Optionally return true when the resource or task is disposed.
|
boolean |
isEmpty()
Deprecated.
|
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) . |
T |
poll()
Deprecated.
|
void |
request(long n)
Deprecated.
|
int |
requestFusion(int requestedMode)
Deprecated.
Request a specific fusion mode from this QueueSubscription.
|
Object |
scanUnsafe(Scannable.Attr key)
Deprecated.
This method is used internally by components to define their key-value mappings
in a single place.
|
int |
size()
Deprecated.
|
void |
subscribe(CoreSubscriber<? super T> actual)
Deprecated.
An internal
Publisher.subscribe(Subscriber) that will bypass
Hooks.onLastOperator(Function) pointcut. |
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. |
dispose, hasCompleted, hasError, isSerialized, serialize, serializeAlways, sink, sink, switchOnNext, wrap
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, contextWrite, contextWrite, count, create, create, defaultIfEmpty, defer, deferContextual, 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, 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, subscriberContext, subscriberContext, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnFirst, switchOnFirst, switchOnNext, switchOnNext, tag, take, take, take, take, takeLast, takeUntil, takeUntilOther, takeWhile, 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, 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
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
add, addAll, contains, containsAll, element, iterator, offer, peek, remove, remove, removeAll, retainAll, toArray, toArray
equals, hashCode, parallelStream, removeIf, spliterator, stream
fusionModeName, fusionModeName
public UnicastProcessor(Queue<T> queue, Disposable onTerminate)
@Deprecated public UnicastProcessor(Queue<T> queue, Consumer<? super T> onOverflow, Disposable onTerminate)
@Deprecated public static <E> UnicastProcessor<E> create()
Sinks.many().unicast().onBackpressureBuffer()
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.UnicastProcessor
that will buffer on an internal queue in an
unbounded fashion.E
- the relayed typeFluxProcessor
@Deprecated public static <E> UnicastProcessor<E> create(Queue<E> queue)
Sinks.many().unicast().onBackpressureBuffer(queue)
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.UnicastProcessor
that will buffer on a provided queue in an
unbounded fashion.E
- the relayed typequeue
- the buffering queueFluxProcessor
@Deprecated public static <E> UnicastProcessor<E> create(Queue<E> queue, Disposable endcallback)
Sinks.many().unicast().onBackpressureBuffer(queue, endCallback)
(or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.UnicastProcessor
that will buffer on a provided queue in an
unbounded fashion.E
- the relayed typequeue
- the buffering queueendcallback
- called on any terminal signalFluxProcessor
@Deprecated public static <E> UnicastProcessor<E> create(Queue<E> queue, Consumer<? super E> onOverflow, Disposable endcallback)
Sinks.many().unicast().onBackpressureBuffer(queue, endCallback)
(or the unsafe variant if you're sure about external synchronization). The onOverflow
callback is not
supported anymore. To be removed in 3.5.UnicastProcessor
that will buffer on a provided queue in an
unbounded fashion.E
- the relayed typequeue
- the buffering queueendcallback
- called on any terminal signalonOverflow
- called when queue.offer return false and unicastProcessor is
about to emit onError.FluxProcessor
public int getBufferSize()
FluxProcessor
Integer.MAX_VALUE
getBufferSize
in class FluxProcessor<T,T>
Integer.MAX_VALUE
public Stream<Scannable> inners()
Scannable
Stream
of referenced inners (flatmap, multicast etc)public Object scanUnsafe(Scannable.Attr key)
Scannable
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.
scanUnsafe
in interface Scannable
scanUnsafe
in class FluxProcessor<T,T>
key
- a Scannable.Attr
to resolve for the component.public void onComplete()
onComplete
in interface Subscriber<T>
public Sinks.EmitResult tryEmitComplete()
Sinks.Many
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 #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.
Sinks.EmitResult
, which should be checked to distinguish different possible failuresSubscriber.onComplete()
public void onError(Throwable throwable)
onError
in interface Subscriber<T>
public Sinks.EmitResult tryEmitError(Throwable t)
Sinks.Many
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 #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.
t
- the exception to signal, not nullSinks.EmitResult
, which should be checked to distinguish different possible failuresSubscriber.onError(Throwable)
public void onNext(T t)
onNext
in interface Subscriber<T>
public void emitNext(T value, Sinks.EmitFailureHandler failureHandler)
Sinks.Many
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, ...).
value
- the value to emit, not nullfailureHandler
- the failure handler that allows retrying failed Sinks.EmitResult
.Sinks.Many.tryEmitNext(Object)
,
Subscriber.onNext(Object)
public Sinks.EmitResult tryEmitNext(T t)
Sinks.Many
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 #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, ...).
t
- the value to emit, not nullSinks.EmitResult
, which should be checked to distinguish different possible failuresSubscriber.onNext(Object)
public int currentSubscriberCount()
Sinks.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.
public Flux<T> asFlux()
Sinks.Many
Flux
view of this sink. Every call returns the same instance.Flux
view associated to this Sinks.Many
protected boolean isIdentityProcessor()
FluxProcessor
FluxProcessor<T, T>
isIdentityProcessor
in class FluxProcessor<T,T>
FluxProcessor<T, T>
public void onSubscribe(Subscription s)
CoreSubscriber
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
.
onSubscribe
in interface Subscriber<T>
onSubscribe
in interface CoreSubscriber<T>
public int getPrefetch()
Flux
Flux
getPrefetch
in class Flux<T>
Flux
, -1 if unspecifiedpublic Context currentContext()
CoreSubscriber
Context
from dependent components which can include downstream
operators during subscribing or a terminal Subscriber
.currentContext
in interface CoreSubscriber<T>
currentContext
in class FluxProcessor<T,T>
Context.empty()
public void subscribe(CoreSubscriber<? super T> actual)
Flux
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.
subscribe
in interface CorePublisher<T>
subscribe
in class Flux<T>
actual
- the Subscriber
interested into the published sequenceFlux.subscribe(Subscriber)
public void request(long n)
request
in interface Subscription
public void cancel()
cancel
in interface Subscription
public int size()
size
in interface Collection<T>
public boolean isEmpty()
isEmpty
in interface Collection<T>
public void clear()
clear
in interface Collection<T>
public int requestFusion(int requestedMode)
Fuseable.QueueSubscription
One should request either SYNC, ASYNC or ANY modes (never NONE) and the implementor should return NONE, SYNC or ASYNC (never ANY).
For example, if a source supports only ASYNC fusion but the intermediate operator supports only SYNC fuseable sources, the operator may request SYNC fusion and the source can reject it via NONE, thus the operator can return NONE as well to downstream and the fusion doesn't happen.
requestFusion
in interface Fuseable.QueueSubscription<T>
requestedMode
- the mode requested by the intermediate operatorpublic boolean isDisposed()
Disposable
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.
isDisposed
in interface Disposable
public boolean isTerminated()
FluxProcessor
isTerminated
in class FluxProcessor<T,T>
@Nullable public Throwable getError()
FluxProcessor
getError
in class FluxProcessor<T,T>
public CoreSubscriber<? super T> actual()
public long downstreamCount()
FluxProcessor
Subscriber
or -1 if untracked.downstreamCount
in class FluxProcessor<T,T>
Subscriber
or -1 if untrackedpublic boolean hasDownstreams()
FluxProcessor
Subscriber
is actively subscribedhasDownstreams
in class FluxProcessor<T,T>
Subscriber
is actively subscribedpublic void emitComplete(Sinks.EmitFailureHandler failureHandler)
Sinks.Many
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, ...).
emitComplete
in interface Sinks.Many<T>
failureHandler
- the failure handler that allows retrying failed Sinks.EmitResult
.Sinks.Many.tryEmitComplete()
,
Subscriber.onComplete()
public void emitError(Throwable error, Sinks.EmitFailureHandler failureHandler)
Sinks.Many
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, ...).
emitError
in interface Sinks.Many<T>
error
- the exception to signal, not nullfailureHandler
- the failure handler that allows retrying failed Sinks.EmitResult
.Sinks.Many.tryEmitError(Throwable)
,
Subscriber.onError(Throwable)