Top Description Fields Constructors Methods
reactor.core.publisher

public final Class DirectProcessor<T>

extends FluxProcessor<T, T>
implements DirectInnerContainer<T>
Class Inheritance
All Implemented Interfaces
reactor.core.publisher.DirectInnerContainer
Annotations
@Deprecated
Type Parameters
<T>
the input and output value type
Imports
java.util.Objects, java.util.concurrent.atomic.AtomicReferenceFieldUpdater, java.util.stream.Stream, org.reactivestreams.Subscription, reactor.core.CoreSubscriber, .Exceptions, .Scannable, reactor.core.publisher.SinkManyBestEffort.DirectInner, .Sinks.EmitResult, reactor.util.annotation.Nullable, reactor.util.context.Context

Deprecated

To be removed in 3.5, prefer clear cut usage of Sinks. Closest sink is Sinks.many().multicast().directBestEffort(), except it doesn't terminate overflowing downstreams.

Dispatches onNext, onError and onComplete signals to zero-to-many Subscribers. Please note, that along with multiple consumers, current implementation of DirectProcessor supports multiple producers. However, all producers must produce messages on the same Thread, otherwise Reactive Streams Spec contract is violated.



Note

DirectProcessor does not coordinate backpressure between its Subscribers and the upstream, but consumes its upstream in an unbounded manner. In the case where a downstream Subscriber is not ready to receive items (hasn't requested yet or enough), it will be terminated with an IllegalStateException. Hence in terms of interaction model, DirectProcessor only supports PUSH from the source through the processor to the Subscribers.



Note

If there are no Subscribers, upstream items are dropped and only the terminal events are retained. A terminated DirectProcessor will emit the terminal signal to late subscribers.



Note

The implementation ignores Subscriptions set via onSubscribe.

Field Summary

Modifier and TypeField and Description
pack-priv Throwable
error

private volatile SinkManyBestEffort.DirectInner<T>[]
subscribers

private static final AtomicReferenceFieldUpdater<DirectProcessor<T>, SinkManyBestEffort.DirectInner<T>[]>
SUBSCRIBERS

Constructor Summary

AccessConstructor and Description
pack-priv
DirectProcessor()

Method Summary

Modifier and TypeMethod and Description
public boolean
add(SinkManyBestEffort.DirectInner<T>
the new SinkManyBestEffort.DirectInner to add
s)

Implements reactor.core.publisher.DirectInnerContainer.add.

Add a new SinkManyBestEffort.DirectInner to this publisher.
public static <
Type of processed signals
E> DirectProcessor<E>

Returns:

a fresh processor
create()

Deprecated To be removed in 3.5.
Create a new DirectProcessor
public Context
currentContext()

Overrides reactor.core.publisher.FluxProcessor.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 long
downstreamCount()

Overrides reactor.core.publisher.FluxProcessor.downstreamCount.

Return the number of active Subscriber or -1 if untracked.
private void
emitComplete()

private void
emitError(Throwable error)

private void
emitNext(T value)

public Throwable
getError()

Overrides reactor.core.publisher.FluxProcessor.getError.

Current error if any, default to null
public int
getPrefetch()

Overrides reactor.core.publisher.Flux.getPrefetch.

The prefetch configuration of the Flux
public boolean
hasDownstreams()

Overrides reactor.core.publisher.FluxProcessor.hasDownstreams.

Return true if any Subscriber is actively subscribed
public Stream<? extends Scannable>
inners()

Overrides reactor.core.publisher.FluxProcessor.inners.

Overrides default reactor.core.Scannable.inners.

Return a Stream of referenced inners (flatmap, multicast etc)
protected boolean
isIdentityProcessor()

Overrides reactor.core.publisher.FluxProcessor.isIdentityProcessor.

Return true if FluxProcessor<T, T>
public boolean
isTerminated()

Overrides reactor.core.publisher.FluxProcessor.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(Subscription
the Subscription that allows requesting data via Subscription#request(long)
s)

Implements org.reactivestreams.Subscriber.onSubscribe, reactor.core.CoreSubscriber.onSubscribe.

Implementors should initialize any state used by onNext(Object) before calling Subscription#request(long).
public void
remove(SinkManyBestEffort.DirectInner<T>
the SinkManyBestEffort.DirectInner to remove
s)

Implements reactor.core.publisher.DirectInnerContainer.remove.

Remove an SinkManyBestEffort.DirectInner from this publisher.
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.
private Sinks.EmitResult
tryEmitComplete()

private Sinks.EmitResult
tryEmitError(Throwable t)

private Sinks.EmitResult
tryEmitNext(T t)

Inherited from reactor.core.publisher.FluxProcessor:
disposegetBufferSizehasCompletedhasErrorisSerializedscanUnsafeserializeserializeAlwayssinksinkswitchOnNextwrap

Field Detail

errorback to summary
pack-priv Throwable error
subscribersback to summary
private volatile SinkManyBestEffort.DirectInner<T>[] subscribers
Annotations
@SuppressWarnings:unchecked
SUBSCRIBERSback to summary
private static final AtomicReferenceFieldUpdater<DirectProcessor<T>, SinkManyBestEffort.DirectInner<T>[]> SUBSCRIBERS
Annotations
@SuppressWarnings:rawtypes

Constructor Detail

DirectProcessorback to summary
pack-priv DirectProcessor()

Method Detail

addback to summary
public boolean add(SinkManyBestEffort.DirectInner<T> s)

Implements reactor.core.publisher.DirectInnerContainer.add.

Doc from reactor.core.publisher.DirectInnerContainer.add.

Add a new SinkManyBestEffort.DirectInner to this publisher.

Parameters
s:SinkManyBestEffort.DirectInner<T>

the new SinkManyBestEffort.DirectInner to add

Returns:boolean

true if the inner could be added, false if the publisher cannot accept new subscribers

Annotations
@Override
createback to summary
public static <E> DirectProcessor<E> create()

Deprecated

To be removed in 3.5. Closest sink is Sinks.many().multicast().directBestEffort(), except it doesn't terminate overflowing downstreams.

Create a new DirectProcessor

Parameters
<E>
Type of processed signals
Returns:DirectProcessor<E>

a fresh processor

Annotations
@Deprecated
currentContextback to summary
public Context currentContext()

Overrides reactor.core.publisher.FluxProcessor.currentContext.

Overrides default reactor.core.CoreSubscriber.currentContext.

Implements reactor.core.publisher.ContextHolder.currentContext.

Doc from reactor.core.CoreSubscriber.currentContext.

Request a Context from dependent components which can include downstream operators during subscribing or a terminal org.reactivestreams.Subscriber.

Returns:Context

a resolved context or Context#empty()

Annotations
@Override
downstreamCountback to summary
public long downstreamCount()

Overrides reactor.core.publisher.FluxProcessor.downstreamCount.

Doc from reactor.core.publisher.FluxProcessor.downstreamCount.

Return the number of active Subscriber or -1 if untracked.

Returns:long

the number of active Subscriber or -1 if untracked

Annotations
@Override
emitCompleteback to summary
private void emitComplete()
emitErrorback to summary
private void emitError(Throwable error)
emitNextback to summary
private void emitNext(T value)
getErrorback to summary
public Throwable getError()

Overrides reactor.core.publisher.FluxProcessor.getError.

Doc from reactor.core.publisher.FluxProcessor.getError.

Current error if any, default to null

Returns:Throwable

Current error if any, default to null

Annotations
@Override
@Nullable
getPrefetchback to summary
public int getPrefetch()

Overrides reactor.core.publisher.Flux.getPrefetch.

Doc from reactor.core.publisher.Flux.getPrefetch.

The prefetch configuration of the Flux

Returns:int

the prefetch configuration of the Flux, -1 if unspecified

Annotations
@Override
hasDownstreamsback to summary
public boolean hasDownstreams()

Overrides reactor.core.publisher.FluxProcessor.hasDownstreams.

Doc from reactor.core.publisher.FluxProcessor.hasDownstreams.

Return true if any Subscriber is actively subscribed

Returns:boolean

true if any Subscriber is actively subscribed

Annotations
@Override
innersback to summary
public Stream<? extends Scannable> inners()

Overrides reactor.core.publisher.FluxProcessor.inners.

Overrides default reactor.core.Scannable.inners.

Doc from reactor.core.Scannable.inners.

Return a Stream of referenced inners (flatmap, multicast etc)

Returns:Stream<? extends Scannable>

a Stream of referenced inners (flatmap, multicast etc)

Annotations
@Override
isIdentityProcessorback to summary
protected boolean isIdentityProcessor()

Overrides reactor.core.publisher.FluxProcessor.isIdentityProcessor.

Doc from reactor.core.publisher.FluxProcessor.isIdentityProcessor.

Return true if FluxProcessor<T, T>

Returns:boolean

true if FluxProcessor<T, T>

Annotations
@Override
isTerminatedback to summary
public boolean isTerminated()

Overrides reactor.core.publisher.FluxProcessor.isTerminated.

Doc from reactor.core.publisher.FluxProcessor.isTerminated.

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

Returns:boolean

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

Annotations
@Override
onCompleteback to summary
public void onComplete()

Implements org.reactivestreams.Subscriber.onComplete.

Doc from org.reactivestreams.Subscriber.onComplete.

Successful terminal state.

No further events will be sent even if Subscription#request(long) is invoked again.

Annotations
@Override
onErrorback to summary
public void onError(Throwable throwable)

Implements org.reactivestreams.Subscriber.onError.

Doc from org.reactivestreams.Subscriber.onError.

Failed terminal state.

No further events will be sent even if Subscription#request(long) is invoked again.

Parameters
throwable:Throwable

the throwable signaled

Annotations
@Override
onNextback to summary
public void onNext(T t)

Implements org.reactivestreams.Subscriber.onNext.

Doc from org.reactivestreams.Subscriber.onNext.

Data notification sent by the Publisher in response to requests to Subscription#request(long).

Parameters
t:T

the element signaled

Annotations
@Override
onSubscribeback to summary
public void onSubscribe(Subscription s)

Implements org.reactivestreams.Subscriber.onSubscribe, reactor.core.CoreSubscriber.onSubscribe.

Doc from reactor.core.CoreSubscriber.onSubscribe.

Implementors should initialize any state used by 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 reactor.util.Logger. Invoked after calling Publisher#subscribe(Subscriber).

No data will start flowing until Subscription#request(long) is invoked.

It is the responsibility of this Subscriber instance to call Subscription#request(long) whenever more data is wanted.

The Publisher will send notifications only in response to Subscription#request(long).

Parameters
s:Subscription

Doc from org.reactivestreams.Subscriber.onSubscribe.

the Subscription that allows requesting data via Subscription#request(long)

Annotations
@Override
removeback to summary
public void remove(SinkManyBestEffort.DirectInner<T> s)

Implements reactor.core.publisher.DirectInnerContainer.remove.

Doc from reactor.core.publisher.DirectInnerContainer.remove.

Remove an SinkManyBestEffort.DirectInner from this publisher. Does nothing if the inner is not currently managed by the publisher.

Parameters
s:SinkManyBestEffort.DirectInner<T>

the SinkManyBestEffort.DirectInner to remove

Annotations
@Override
@SuppressWarnings:unchecked
subscribeback to summary
public void subscribe(CoreSubscriber<? super T> actual)

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

Implements reactor.core.CorePublisher.subscribe.

Doc from reactor.core.publisher.Flux.subscribe.

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.

Parameters
actual:CoreSubscriber<? super T>

the Subscriber interested into the published sequence

Annotations
@Override
tryEmitCompleteback to summary
private Sinks.EmitResult tryEmitComplete()
tryEmitErrorback to summary
private Sinks.EmitResult tryEmitError(Throwable t)
tryEmitNextback to summary
private Sinks.EmitResult tryEmitNext(T t)