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:

org.reactivestreams.Processor<T,T>, org.reactivestreams.Publisher<T>, org.reactivestreams.Subscriber<T>, Fuseable, MultiProducer, Receiver, Trackable

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

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>

Field Summary

Fields inherited from interface reactor.core.Fuseable

ANY, ASYNC, NONE, SYNC, THREAD_BARRIER

Fields inherited from interface reactor.core.Trackable

UNSPECIFIED

Method Summary

Modifier and Type	Method and Description
static <T> ReplayProcessor<T>	cacheLast() Create a ReplayProcessor from hot-cold ReplayProcessor that will not propagate cancel upstream if Subscription has been set.
static <T> ReplayProcessor<T>	cacheLastOrDefault(T value) Create a ReplayProcessor from hot-cold ReplayProcessor that will not propagate cancel upstream if Subscription has been set.
ReplayProcessor<T>	connect() Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving onNext, onComplete and onError calls.
static <E> ReplayProcessor<E>	create() Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> ReplayProcessor<E>	create(int historySize) Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> ReplayProcessor<E>	create(int historySize, boolean unbounded) Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <T> ReplayProcessor<T>	createSizeAndTimeout(int size, java.time.Duration maxAge) Creates a time- and size-bounded replay processor.
static <T> ReplayProcessor<T>	createSizeAndTimeoutMillis(int size, long maxAge, TimedScheduler scheduler) Creates a time- and size-bounded replay processor.
static <T> ReplayProcessor<T>	createTimeout(java.time.Duration maxAge) Creates a time-bounded replay processor.
static <T> ReplayProcessor<T>	createTimeoutMillis(long maxAge, TimedScheduler scheduler) Creates a time-bounded replay processor.
long	downstreamCount() the number of downstream receivers
java.util.Iterator<?>	downstreams() the connected data receivers
long	getCapacity() Return defined element capacity
long	getPrefetch() The prefetch configuration of the Flux
boolean	isStarted() Has this upstream started or "onSubscribed" ?
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
void	onComplete()
void	onError(java.lang.Throwable t)
void	onNext(T t)
void	onSubscribe(org.reactivestreams.Subscription s)
void	subscribe(org.reactivestreams.Subscriber<? super T> s)
java.lang.Object	upstream() Return the direct source of data, Supports reference.

Methods inherited from class reactor.core.publisher.FluxProcessor

connectSink, connectSink, serialize, switchOnNext, wrap

Methods inherited from class reactor.core.publisher.Flux

all, any, as, awaitOnSubscribe, blockFirst, blockFirst, blockFirstMillis, blockLast, blockLast, blockLastMillis, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, cache, cache, cache, cache, cancelOn, cast, collect, collect, collectList, collectMap, collectMap, collectMap, collectMultimap, collectMultimap, collectMultimap, collectSortedList, collectSortedList, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, compose, concat, concat, concat, concat, concatDelayError, concatDelayError, concatDelayError, concatDelayError, concatMap, concatMap, concatMapDelayError, concatMapDelayError, concatMapDelayError, concatMapIterable, concatMapIterable, concatWith, count, create, create, defaultIfEmpty, defer, delay, delayMillis, delayMillis, delaySubscription, delaySubscription, delaySubscriptionMillis, delaySubscriptionMillis, dematerialize, distinct, distinct, distinctUntilChanged, distinctUntilChanged, doAfterTerminate, doOnCancel, doOnComplete, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, error, error, filter, firstEmitting, firstEmitting, firstEmittingWith, flatMap, flatMap, flatMap, flatMap, flatMap, flatMapIterable, flatMapIterable, from, fromArray, fromIterable, fromStream, generate, generate, generate, groupBy, groupBy, groupJoin, handle, hasElement, hasElements, hide, ignoreElements, interval, interval, intervalMillis, intervalMillis, intervalMillis, intervalMillis, join, just, just, last, last, log, log, log, log, map, mapError, mapError, mapError, materialize, merge, merge, merge, merge, merge, merge, mergeWith, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorResumeWith, onErrorResumeWith, onErrorResumeWith, onErrorReturn, onErrorReturn, onErrorReturn, onTerminateDetach, parallel, parallel, parallel, publish, publish, publish, publish, publishNext, publishOn, publishOn, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, replay, replay, replayMillis, replayMillis, retry, retry, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleFirstMillis, sampleMillis, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, single, single, singleOrEmpty, skip, skip, skipLast, skipMillis, skipMillis, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnError, switchOnError, switchOnError, switchOnNext, switchOnNext, take, take, takeLast, takeMillis, takeMillis, takeUntil, takeUntilOther, takeWhile, then, then, then, thenMany, thenMany, timeout, timeout, timeout, timeout, timeout, timeoutMillis, timeoutMillis, timeoutMillis, timeoutMillis, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, toString, transform, using, using, window, window, window, window, window, window, window, window, windowMillis, windowMillis, windowMillis, windowMillis, windowMillis, withLatestFrom, 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 reactor.core.MultiProducer

hasDownstreams

Methods inherited from interface reactor.core.Trackable

expectedFromUpstream, getError, getPending, isCancelled, limit, requestedFromDownstream

Method Detail

cacheLast

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

Create a ReplayProcessor from hot-cold ReplayProcessor that will not propagate cancel upstream if Subscription has been set. The last emitted item will be replayable to late Subscriber (buffer and history size of 1).

Type Parameters:

T - the relayed type

Returns:

a non interruptable last item cached pub-sub ReplayProcessor

cacheLastOrDefault

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

Create a ReplayProcessor from hot-cold ReplayProcessor that will not propagate cancel upstream if Subscription has been set. The last emitted item will be replayable to late Subscriber (buffer and history size of 1).

Type Parameters:

T - the relayed type

Parameters:

value - a default value to start the sequence with

Returns:

a non interruptable last item cached pub-sub ReplayProcessor

create

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

Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

create

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

Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

historySize -

Returns:

a fresh processor

create

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

Create a new ReplayProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

Type Parameters:

E - Type of processed signals

Parameters:

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

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

Returns:

a fresh processor

createTimeout

public static <T> ReplayProcessor<T> createTimeout(java.time.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.timer() 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

createTimeoutMillis

public static <T> ReplayProcessor<T> createTimeoutMillis(long maxAge,
                                                         TimedScheduler scheduler)

Creates a time-bounded replay processor.

In this setting, the ReplayProcessor internally tags each observed item with a timestamp value supplied by the TimedScheduler 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 in milliseconds

scheduler - the TimedScheduler that provides the current time

Returns:

a new ReplayProcessor

createSizeAndTimeout

public static <T> ReplayProcessor<T> createSizeAndTimeout(int size,
                                                          java.time.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.timer() 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

createSizeAndTimeoutMillis

public static <T> ReplayProcessor<T> createSizeAndTimeoutMillis(int size,
                                                                long maxAge,
                                                                TimedScheduler 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 TimedScheduler 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 TimedScheduler that provides the current time

Returns:

a new ReplayProcessor

subscribe

public void subscribe(org.reactivestreams.Subscriber<? super T> s)

Specified by:

subscribe in interface org.reactivestreams.Publisher<T>

Overrides:

subscribe in class FluxProcessor<T,T>

downstreams

public java.util.Iterator<?> downstreams()

Description copied from interface: MultiProducer

the connected data receivers

Specified by:

downstreams in interface MultiProducer

Returns:

the connected data receivers

downstreamCount

public long downstreamCount()

Description copied from interface: MultiProducer

the number of downstream receivers

Specified by:

downstreamCount in interface MultiProducer

Returns:

the number of downstream receivers

getCapacity

public long getCapacity()

Description copied from interface: Trackable

Return defined element capacity

Specified by:

getCapacity in interface Trackable

Overrides:

getCapacity in class FluxProcessor<T,T>

Returns:

long capacity

isTerminated

public boolean isTerminated()

Description copied from interface: Trackable

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

Specified by:

isTerminated in interface Trackable

Returns:

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

isStarted

public boolean isStarted()

Description copied from interface: Trackable

Has this upstream started or "onSubscribed" ?

Specified by:

isStarted in interface Trackable

Returns:

has this upstream started or "onSubscribed" ?

upstream

public java.lang.Object upstream()

Description copied from interface: Receiver

Return the direct source of data, Supports reference.

Specified by:

upstream in interface Receiver

Returns:

the direct source of data, Supports reference.

onSubscribe

public void onSubscribe(org.reactivestreams.Subscription s)

Specified by:

onSubscribe in interface org.reactivestreams.Subscriber<T>

getPrefetch

public long 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, -1L if unspecified

onNext

public void onNext(T t)

Specified by:

onNext in interface org.reactivestreams.Subscriber<T>

onError

public void onError(java.lang.Throwable t)

Specified by:

onError in interface org.reactivestreams.Subscriber<T>

onComplete

public void onComplete()

Specified by:

onComplete in interface org.reactivestreams.Subscriber<T>

connect

public ReplayProcessor<T> connect()

Description copied from class: FluxProcessor

Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving onNext, onComplete and onError calls.

Doing so MAY allow direct UNBOUNDED onXXX calls and MAY prevent Publisher to subscribe this subscriber. Note that Processor can extend this behavior to effectively start its subscribers.

Overrides:

connect in class FluxProcessor<T,T>

Returns:

this