reactor.core.publisher

Class ReplayProcessor<T>

java.lang.Object

reactor.core.publisher.Flux<OUT>

reactor.core.publisher.FluxProcessor<T,T>

reactor.core.publisher.ReplayProcessor<T>

Type Parameters:

T - the value type

All Implemented Interfaces:

Processor<T,T>, Publisher<T>, Subscriber<T>, CorePublisher<T>, CoreSubscriber<T>, Disposable, Fuseable, Scannable

public final class ReplayProcessor<T>
extends FluxProcessor<T,T>
implements Fuseable

Replays all or the last N items to Subscribers.

Nested Class Summary

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

Fuseable.ConditionalSubscriber<T>, Fuseable.QueueSubscription<T>, Fuseable.ScalarCallable<T>, Fuseable.SynchronousSubscription<T>

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

Scannable.Attr<T>

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

Disposable.Composite, Disposable.Swap

Field Summary

Fields inherited from interface reactor.core.Fuseable

ANY, ASYNC, NONE, SYNC, THREAD_BARRIER

Fields inherited from interface reactor.core.Scannable

OPERATOR_NAME_UNRELATED_WORDS_PATTERN

Method Summary

Modifier and Type	Method and Description
static <T> ReplayProcessor<T>	cacheLast() Create a ReplayProcessor that caches the last element it has pushed, replaying it to late subscribers.
static <T> ReplayProcessor<T>	cacheLastOrDefault(T value) Create a ReplayProcessor that caches the last element it has pushed, replaying it to late subscribers.
static <E> ReplayProcessor<E>	create() Create a new ReplayProcessor that replays an unbounded number of elements, using a default internal Queue.
static <E> ReplayProcessor<E>	create(int historySize) Create a new ReplayProcessor that replays up to historySize elements.
static <E> ReplayProcessor<E>	create(int historySize, boolean unbounded) Create a new ReplayProcessor that either replay all the elements or a limited amount of elements depending on the unbounded parameter.
static <T> ReplayProcessor<T>	createSizeAndTimeout(int size, Duration maxAge) Creates a time- and size-bounded replay processor.
static <T> ReplayProcessor<T>	createSizeAndTimeout(int size, Duration maxAge, Scheduler scheduler) Creates a time- and size-bounded replay processor.
static <T> ReplayProcessor<T>	createTimeout(Duration maxAge) Creates a time-bounded replay processor.
static <T> ReplayProcessor<T>	createTimeout(Duration maxAge, Scheduler scheduler) Creates a time-bounded replay processor.
Context	currentContext() Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
long	downstreamCount() Return the number of active Subscriber or -1 if untracked.
Throwable	getError() Current error if any, default to null
int	getPrefetch() The prefetch configuration of the Flux
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
void	onComplete()
void	onError(Throwable t)
void	onNext(T t)
void	onSubscribe(Subscription s) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
Object	scanUnsafe(Scannable.Attr key) This method is used internally by components to define their key-value mappings in a single place.
void	subscribe(CoreSubscriber<? super T> actual) An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.

Methods inherited from class reactor.core.publisher.FluxProcessor

dispose, getBufferSize, 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, 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, count, create, create, defaultIfEmpty, defer, 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, 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, materialize, merge, merge, merge, merge, merge, merge, mergeDelayError, mergeOrdered, mergeOrdered, mergeOrdered, mergeOrderedWith, 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, 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, retry, retry, retryBackoff, retryBackoff, retryBackoff, retryBackoff, retryBackoff, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, 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, switchOnNext, switchOnNext, tag, take, take, take, takeLast, takeUntil, takeUntilOther, takeWhile, then, then, thenEmpty, thenMany, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, toString, transform, transformDeferred, using, using, usingWhen, usingWhen, window, window, window, window, window, window, window, 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, from, isScanAvailable, name, parents, scan, scanOrDefault, stepName, steps, tags

Methods inherited from interface reactor.core.Disposable

isDisposed

Method Detail

cacheLast

public static <T> ReplayProcessor<T> cacheLast()

Create a ReplayProcessor that caches the last element it has pushed, replaying it to late subscribers. This is a buffer-based ReplayProcessor with a history size of 1.

Type Parameters:

T - the type of the pushed elements

Returns:

a new ReplayProcessor that replays its last pushed element to each new Subscriber

cacheLastOrDefault

public static <T> ReplayProcessor<T> cacheLastOrDefault(@Nullable
                                                        T value)

Create a ReplayProcessor that caches the last element it has pushed, replaying it to late subscribers. If a Subscriber comes in before any value has been pushed, then the defaultValue is emitted instead. This is a buffer-based ReplayProcessor with a history size of 1.

Type Parameters:

T - the type of the pushed elements

Parameters:

value - a default value to start the sequence with in case nothing has been cached yet.

Returns:

a new ReplayProcessor that replays its last pushed element to each new Subscriber, or a default one if nothing was pushed yet

create

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

Create a new ReplayProcessor that replays an unbounded number of elements, using a default internal Queue.

Type Parameters:

E - the type of the pushed elements

Returns:

a new ReplayProcessor that replays the whole history to each new Subscriber.

create

public static <E> ReplayProcessor<E> create(int historySize)

Create a new ReplayProcessor that replays up to historySize elements.

Type Parameters:

E - the type of the pushed elements

Parameters:

historySize - the backlog size, ie. maximum items retained for replay.

Returns:

a new ReplayProcessor that replays a limited history to each new Subscriber.

create

public static <E> ReplayProcessor<E> create(int historySize,
                                            boolean unbounded)

Create a new ReplayProcessor that either replay all the elements or a limited amount of elements depending on the unbounded parameter.

Type Parameters:

E - the type of the pushed elements

Parameters:

historySize - maximum items retained if bounded, or initial link size if unbounded

unbounded - true if "unlimited" data store must be supplied

Returns:

a new ReplayProcessor that replays the whole history to each new Subscriber if configured as unbounded, a limited history otherwise.

createSizeAndTimeout

public static <T> ReplayProcessor<T> createSizeAndTimeout(int size,
                                                          Duration maxAge)

Creates a time- and size-bounded replay processor.

In this setting, the ReplayProcessor internally tags each received item with a timestamp value supplied by the Schedulers.parallel() and holds at most size items in its internal buffer. It evicts items from the start of the buffer if their age becomes less-than or equal to the supplied age in milliseconds or the buffer reaches its size limit.

When subscribers subscribe to a terminated ReplayProcessor, they observe the items that remained in the buffer after the terminal signal, regardless of their age, but at most size items.

If an subscriber subscribes while the ReplayProcessor is active, it will observe only those items from within the buffer that have age less than the specified time and each subsequent item, even if the buffer evicts items due to the time constraint in the mean time. In other words, once an subscriber subscribes, it observes items without gaps in the sequence except for the outdated items at the beginning of the sequence.

Note that terminal signals (onError and onComplete) trigger eviction as well. For example, with a max age of 5, the first item is observed at T=0, then an onComplete signal arrives at T=10. If an Subscriber subscribes at T=11, it will find an empty ReplayProcessor with just an onCompleted signal.

Type Parameters:

T - the type of items observed and emitted by the Processor

Parameters:

maxAge - the maximum age of the contained items

size - the maximum number of buffered items

Returns:

a new ReplayProcessor that replay up to size elements, but will evict them from its history based on their age.

createSizeAndTimeout

public static <T> ReplayProcessor<T> createSizeAndTimeout(int size,
                                                          Duration maxAge,
                                                          Scheduler scheduler)

Creates a time- and size-bounded replay processor.

In this setting, the ReplayProcessor internally tags each received item with a timestamp value supplied by the Scheduler and holds at most size items in its internal buffer. It evicts items from the start of the buffer if their age becomes less-than or equal to the supplied age in milliseconds or the buffer reaches its size limit.

When subscribers subscribe to a terminated ReplayProcessor, they observe the items that remained in the buffer after the terminal signal, regardless of their age, but at most size items.

If an subscriber subscribes while the ReplayProcessor is active, it will observe only those items from within the buffer that have age less than the specified time and each subsequent item, even if the buffer evicts items due to the time constraint in the mean time. In other words, once an subscriber subscribes, it observes items without gaps in the sequence except for the outdated items at the beginning of the sequence.

Note that terminal signals (onError and onComplete) trigger eviction as well. For example, with a max age of 5, the first item is observed at T=0, then an onComplete signal arrives at T=10. If an Subscriber subscribes at T=11, it will find an empty ReplayProcessor with just an onCompleted signal.

Type Parameters:

T - the type of items observed and emitted by the Processor

Parameters:

maxAge - the maximum age of the contained items in milliseconds

size - the maximum number of buffered items

scheduler - the Scheduler that provides the current time

Returns:

a new ReplayProcessor that replay up to size elements, but will evict them from its history based on their age.

createTimeout

public static <T> ReplayProcessor<T> createTimeout(Duration maxAge)

Creates a time-bounded replay processor.

In this setting, the ReplayProcessor internally tags each observed item with a timestamp value supplied by the Schedulers.parallel() and keeps only those whose age is less than the supplied time value converted to milliseconds. For example, an item arrives at T=0 and the max age is set to 5; at T>=5 this first item is then evicted by any subsequent item or termination signal, leaving the buffer empty.

Once the processor is terminated, subscribers subscribing to it will receive items that remained in the buffer after the terminal signal, regardless of their age.

If an subscriber subscribes while the ReplayProcessor is active, it will observe only those items from within the buffer that have an age less than the specified time, and each item observed thereafter, even if the buffer evicts items due to the time constraint in the mean time. In other words, once an subscriber subscribes, it observes items without gaps in the sequence except for any outdated items at the beginning of the sequence.

Note that terminal signals (onError and onComplete) trigger eviction as well. For example, with a max age of 5, the first item is observed at T=0, then an onComplete signal arrives at T=10. If an subscriber subscribes at T=11, it will find an empty ReplayProcessor with just an onCompleted signal.

Type Parameters:

T - the type of items observed and emitted by the Processor

Parameters:

maxAge - the maximum age of the contained items

Returns:

a new ReplayProcessor that replays elements based on their age.

createTimeout

public static <T> ReplayProcessor<T> createTimeout(Duration maxAge,
                                                   Scheduler scheduler)

Creates a time-bounded replay processor.

In this setting, the ReplayProcessor internally tags each observed item with a timestamp value supplied by the Scheduler and keeps only those whose age is less than the supplied time value converted to milliseconds. For example, an item arrives at T=0 and the max age is set to 5; at T>=5 this first item is then evicted by any subsequent item or termination signal, leaving the buffer empty.

Once the processor is terminated, subscribers subscribing to it will receive items that remained in the buffer after the terminal signal, regardless of their age.

If an subscriber subscribes while the ReplayProcessor is active, it will observe only those items from within the buffer that have an age less than the specified time, and each item observed thereafter, even if the buffer evicts items due to the time constraint in the mean time. In other words, once an subscriber subscribes, it observes items without gaps in the sequence except for any outdated items at the beginning of the sequence.

Note that terminal signals (onError and onComplete) trigger eviction as well. For example, with a max age of 5, the first item is observed at T=0, then an onComplete signal arrives at T=10. If an subscriber subscribes at T=11, it will find an empty ReplayProcessor with just an onCompleted signal.

Type Parameters:

T - the type of items observed and emitted by the Processor

Parameters:

maxAge - the maximum age of the contained items

Returns:

a new ReplayProcessor that replays elements based on their age.

currentContext

public Context currentContext()

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>

Returns:

a resolved context or Context.empty()

downstreamCount

public long downstreamCount()

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

getError

@Nullable
public Throwable getError()

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

getPrefetch

public int getPrefetch()

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

inners

public Stream<? extends Scannable> inners()

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)

isTerminated

public boolean isTerminated()

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" ?

onComplete

public void onComplete()

Specified by:

onComplete in interface Subscriber<T>

onError

public void onError(Throwable t)

Specified by:

onError in interface Subscriber<T>

onNext

public void onNext(T t)

Specified by:

onNext in interface Subscriber<T>

onSubscribe

public void onSubscribe(Subscription s)

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 Subscriber<T>

Specified by:

onSubscribe in interface CoreSubscriber<T>

scanUnsafe

@Nullable
public Object scanUnsafe(Scannable.Attr key)

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.

subscribe

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

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)