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

Constructor Summary

Constructors

Constructor and Description
ReplayProcessor(int bufferSize, boolean unbounded) Constructs a ReplayProcessor with bounded or unbounded buffering.

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.
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)
static <T> FluxProcessor<T,T>	serialize() Create a FluxProcessor from hot ReplayProcessor safely gated by a serializing Subscriber.
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, 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, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, cache, cache, 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, elementAt, elementAtOrDefault, empty, error, error, every, everyFirst, 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, log, log, log, map, mapError, mapError, mapError, materialize, merge, merge, merge, merge, merge, merge, mergeWith, never, next, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorResumeWith, onErrorResumeWith, onErrorResumeWith, onErrorReturn, onErrorReturn, onErrorReturn, onTerminateDetach, parallel, parallel, parallel, process, process, process, process, publish, publish, publish, publish, publishNext, publishOn, publishOn, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, retry, retry, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleFirstMillis, sampleMillis, sampleTimeout, sampleTimeout, scan, scan, scanWith, single, singleOrDefault, singleOrEmpty, skip, skip, skipLast, skipMillis, skipMillis, skipUntil, skipUntilOther, skipWhile, 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, toIterable, toIterable, toIterable, toStream, toStream, toString, transform, using, using, window, window, window, window, window, window, window, window, windowMillis, 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

Constructor Detail

ReplayProcessor

public ReplayProcessor(int bufferSize,
                       boolean unbounded)

Constructs a ReplayProcessor with bounded or unbounded buffering.

Parameters:

bufferSize - if unbounded, this number represents the link size of the shared buffer, if bounded, this is the maximum number of retained items

unbounded - should the replay buffer be unbounded

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

serialize

public static <T> FluxProcessor<T,T> serialize()

Create a FluxProcessor from hot ReplayProcessor safely gated by a serializing Subscriber. It will not propagate cancel upstream if Subscription has been set. Serialization uses thread-stealing and a potentially unbounded queue that might starve a calling thread if races are too important and Subscriber is slower.

Type Parameters:

T - the relayed type

Returns:

a serializing FluxProcessor

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