Top Description Inners Fields Constructors Methods
reactor.core.publisher

pack-priv final Class SinkManyEmitterProcessor<T>

extends Flux<T>
implements InternalManySink<T>, Sinks.ManyWithUpstream<T>, CoreSubscriber<T>, Scannable, Disposable, ContextHolder
Class Inheritance
All Implemented Interfaces
reactor.core.publisher.ContextHolder, reactor.core.Disposable, reactor.core.Scannable, reactor.core.CoreSubscriber, org.reactivestreams.Subscriber, reactor.core.publisher.Sinks.ManyWithUpstream, reactor.core.publisher.Sinks.Many, reactor.core.publisher.InternalManySink
Type Parameters
<T>
the input and output value type
Imports
java.util.Objects, .Queue, java.util.concurrent.CancellationException, java.util.concurrent.atomic.AtomicIntegerFieldUpdater, .AtomicReferenceFieldUpdater, java.util.stream.Stream, org.reactivestreams.Publisher, .Subscriber, .Subscription, reactor.core.CoreSubscriber, .Disposable, .Exceptions, .Fuseable, .Scannable, reactor.core.publisher.Sinks.EmitResult, reactor.util.annotation.Nullable, reactor.util.concurrent.Queues, reactor.util.context.Context
An implementation of a Sinks.ManyWithUpstream 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 and Inner Type Summary

Modifier and TypeClass and Description
pack-priv static class
SinkManyEmitterProcessor.EmitterDisposable

pack-priv static class
SinkManyEmitterProcessor.EmitterInner<T>

Field Summary

Modifier and TypeField and Description
pack-priv final boolean
autoCancel

pack-priv volatile boolean
done

pack-priv static final FluxPublish.PubSubInner<T>[]
EMPTY

pack-priv volatile Throwable
error

pack-priv static final AtomicReferenceFieldUpdater<SinkManyEmitterProcessor<T>, Throwable>
ERROR

pack-priv final int
prefetch

pack-priv volatile Queue<T>
queue

pack-priv volatile Subscription
s

pack-priv static final AtomicReferenceFieldUpdater<SinkManyEmitterProcessor<T>, Subscription>
S

pack-priv int
sourceMode

pack-priv volatile FluxPublish.PubSubInner<T>[]
subscribers

pack-priv static final AtomicReferenceFieldUpdater<SinkManyEmitterProcessor<T>, FluxPublish.PubSubInner<T>[]>
SUBSCRIBERS

pack-priv static final AtomicReferenceFieldUpdater<SinkManyEmitterProcessor<T>, SinkManyEmitterProcessor.EmitterDisposable>
UPSTREAM_DISPOSABLE

pack-priv volatile SinkManyEmitterProcessor.EmitterDisposable
upstreamDisposable

pack-priv volatile int
wip

pack-priv static final AtomicIntegerFieldUpdater<SinkManyEmitterProcessor<T>>
WIP

Inherited from reactor.core.publisher.Flux:
ALWAYS_BOOLEAN_SUPPLIERIDENTITY_FUNCTIONLIST_SUPPLIEROBJECT_EQUALSET_SUPPLIERTUPLE2_BIFUNCTION

Constructor Summary

AccessConstructor and Description
pack-priv
SinkManyEmitterProcessor(boolean autoCancel, int prefetch)

Method Summary

Modifier and TypeMethod and Description
pack-priv final boolean
add(SinkManyEmitterProcessor.EmitterInner<T> inner)

public Flux<T>
asFlux()

Implements reactor.core.publisher.Sinks.Many.asFlux.

Return a Flux view of this sink.
pack-priv boolean
checkTerminated(boolean d, boolean empty)

public Context
currentContext()

Overrides default reactor.core.CoreSubscriber.currentContext.

Implements reactor.core.publisher.ContextHolder.currentContext.

Request a Context from dependent components which can include downstream operators during subscribing or a terminal org.reactivestreams.Subscriber.
public int
currentSubscriberCount()

Implements reactor.core.publisher.Sinks.Many.currentSubscriberCount.

Get how many Subscribers are currently subscribed to the sink.
private boolean
detach()

public void
dispose()

Implements reactor.core.Disposable.dispose.

Cancel or dispose the underlying task or resource.
pack-priv final void
drain()

pack-priv Throwable

Returns:

Current error if any, default to null
getError()

Current error if any, default to null

pack-priv int

Returns:

the number of parked elements in the emitter backlog.
getPending()

Return the number of parked elements in the emitter backlog.

public int
getPrefetch()

Overrides reactor.core.publisher.Flux.getPrefetch.

The prefetch configuration of the Flux
public Stream<? extends Scannable>
inners()

Overrides default reactor.core.Scannable.inners.

Return a Stream of referenced inners (flatmap, multicast etc)
pack-priv boolean

Returns:

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

private boolean
isDetached()

public boolean
isDisposed()

Overrides default reactor.core.Disposable.isDisposed.

Optionally return true when the resource or task is disposed.
public boolean

Returns:

has this upstream finished or "completed" / "failed" ?
isTerminated()

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

public void
onComplete()

Implements org.reactivestreams.Subscriber.onComplete.

Successful terminal state.
public void
onError(Throwable
the throwable signaled
throwable)

Implements org.reactivestreams.Subscriber.onError.

Failed terminal state.
public void
onNext(T
the element signaled
t)

Implements org.reactivestreams.Subscriber.onNext.

Data notification sent by the Publisher in response to requests to Subscription#request(long).
public void
onSubscribe(final Subscription
the Subscription that allows requesting data via Subscription#request(long)
s)

Implements reactor.core.CoreSubscriber.onSubscribe.

Implementors should initialize any state used by onNext(Object) before calling Subscription#request(long).
pack-priv final void
remove(FluxPublish.PubSubInner<T> inner)

public Object
scanUnsafe(Scannable.Attr<T>
a Attr to resolve for the component.
key)

Implements reactor.core.Scannable.scanUnsafe.

This method is used internally by components to define their key-value mappings in a single place.
public void
subscribe(CoreSubscriber<? super T>
the Subscriber interested into the published sequence
actual)

Implements abstract reactor.core.publisher.Flux.subscribe.

Implements reactor.core.CorePublisher.subscribe.

An internal Publisher#subscribe(Subscriber) that will bypass Hooks#onLastOperator(Function) pointcut.
public Disposable
subscribeTo(Publisher<? extends T> upstream)

Implements reactor.core.publisher.Sinks.ManyWithUpstream.subscribeTo.

Explicitly subscribe this Sinks.Many to an upstream Publisher without exposing it as a Subscriber at all.
pack-priv FluxPublish.PubSubInner<T>[]
terminate()

public Sinks.EmitResult
tryEmitComplete()

Implements reactor.core.publisher.Sinks.Many.tryEmitComplete.

Try to terminate the sequence successfully, generating an onComplete signal.
public Sinks.EmitResult
tryEmitError(Throwable
the exception to signal, not null
t)

Implements reactor.core.publisher.Sinks.Many.tryEmitError.

Try to fail the sequence, generating an onError signal.
public Sinks.EmitResult
tryEmitNext(T
the value to emit, not null
t)

Implements reactor.core.publisher.Sinks.Many.tryEmitNext.

Try emitting a non-null element, generating an onNext signal.
Inherited from reactor.core.publisher.Flux:
allanyasblockFirstblockFirstblockLastblockLastbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferUntilbufferUntilbufferUntilChangedbufferUntilChangedbufferUntilChangedbufferWhenbufferWhenbufferWhilecachecachecachecachecachecachecancelOncastcheckpointcheckpointcheckpointcollectcollectcollectListcollectMapcollectMapcollectMapcollectMultimapcollectMultimapcollectMultimapcollectSortedListcollectSortedListcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestconcatconcatconcatconcatconcatDelayErrorconcatDelayErrorconcatDelayErrorconcatDelayErrorconcatMapconcatMapconcatMapDelayErrorconcatMapDelayErrorconcatMapDelayErrorconcatMapIterableconcatMapIterableconcatWithconcatWithValuescontextCapturecontextWritecontextWritecountcountingBooleanSuppliercountingPredicatecreatecreatedefaultIfEmptydeferdeferContextualdelayElementsdelayElementsdelaySequencedelaySequencedelaySubscriptiondelaySubscriptiondelaySubscriptiondelayUntildematerializedistinctdistinctdistinctdistinctdistinctUntilChangeddistinctUntilChangeddistinctUntilChangeddoAfterTerminatedoFinallydoFirstdoOnCanceldoOnCompletedoOnDiscarddoOnEachdoOnErrordoOnErrordoOnErrordoOnNextdoOnRequestdoOnSignaldoOnSubscribedoOnTerminateelapsedelapsedelementAtelementAtemptyequalPredicateerrorerrorerrorexpandexpandexpandDeepexpandDeepfilterfilterWhenfilterWhenfirstfirstfirstWithSignalfirstWithSignalfirstWithValuefirstWithValueflatMapflatMapflatMapflatMapflatMapflatMapDelayErrorflatMapIterableflatMapIterableflatMapSequentialflatMapSequentialflatMapSequentialflatMapSequentialflatMapSequentialDelayErrorfromfromArrayfromIterablefromStreamfromStreamgenerategenerategenerategroupBygroupBygroupBygroupBygroupJoinhandlehasElementhasElementshashSetSupplierhideidentityFunctionignoreElementsindexindexintervalintervalintervalintervaljoinjustjustlastlastlimitRatelimitRatelimitRequestlistSupplierloglogloglogloglogmapmapNotNullmaterializemergemergemergemergemergemergemergemergeComparingmergeComparingmergeComparingmergeComparingDelayErrormergeComparingWithmergeDelayErrormergeOrderedmergeOrderedmergeOrderedmergeOrderedWithmergePrioritymergePrioritymergePrioritymergePriorityDelayErrormergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialDelayErrormergeSequentialDelayErrormergeSequentialDelayErrormergeWithmetricsnamenevernextofTypeonAssemblyonAssemblyonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureDroponBackpressureDroponBackpressureErroronBackpressureLatestonErrorCompleteonErrorCompleteonErrorCompleteonErrorContinueonErrorContinueonErrorContinueonErrorMaponErrorMaponErrorMaponErrorResumeonErrorResumeonErrorResumeonErrorReturnonErrorReturnonErrorReturnonErrorStoponTerminateDetachorparallelparallelparallelpublishpublishpublishpublishpublishNextpublishOnpublishOnpublishOnpublishOnpushpushrangereducereducereduceWithrepeatrepeatrepeatrepeatrepeatWhenreplayreplayreplayreplayreplayreplayretryretryretryWhensamplesamplesampleFirstsampleFirstsampleTimeoutsampleTimeoutscanscanscanWithshareshareNextsinglesinglesingleOrEmptyskipskipskipskipLastskipUntilskipUntilOtherskipWhilesortsortstartWithstartWithstartWithsubscribesubscribesubscribesubscribesubscribesubscribesubscribesubscribeOnsubscribeOnsubscribeWithswitchIfEmptyswitchMapswitchMapswitchOnFirstswitchOnFirstswitchOnNextswitchOnNexttagtaketaketaketaketakeLasttakeUntiltakeUntilOthertakeWhiletaptaptapthenthenthenEmptythenManytimedtimedtimeouttimeouttimeouttimeouttimeouttimeouttimeouttimestamptimestamptoIterabletoIterabletoIterabletoStreamtoStreamtoStringtransformtransformDeferredtransformDeferredContextualtuple2FunctionusingusingusingWhenusingWhenwindowwindowwindowwindowwindowwindowwindowwindowTimeoutwindowTimeoutwindowTimeoutwindowTimeoutwindowUntilwindowUntilwindowUntilwindowUntilChangedwindowUntilChangedwindowUntilChangedwindowWhenwindowWhilewindowWhilewithLatestFromwrapwrapToMonozipzipzipzipzipzipzipzipzipzipzipzipzipzipWithzipWithzipWithzipWithzipWithIterablezipWithIterable