public final Class ReplayProcessor<T>

The ReplayProcessor supports the following item retainment strategies:

• create() and create(int): retains and replays all events to current and future Subscribers.

  

  

• createWithSize(int): retains at most the given number of items and replays only these latest items to new Subscribers.

  

• createWithTime(long, TimeUnit, Scheduler): retains items no older than the specified time and replays them to new Subscribers (which could mean all items age out).

  

• createWithTimeAndSize(long, TimeUnit, Scheduler, int): retains no more than the given number of items which are also no older than the specified time and replays them to new Subscribers (which could mean all items age out).

  

The ReplayProcessor can be created in bounded and unbounded mode. It can be bounded by size (maximum number of elements retained at most) and/or time (maximum age of elements replayed).

Since a ReplayProcessor is a Reactive Streams Processor, nulls are not allowed (Rule 2.13) as parameters to onNext(Object) and onError(Throwable). Such calls will result in a NullPointerException being thrown and the processor's state is not changed.

This ReplayProcessor respects the individual backpressure behavior of its Subscribers but does not coordinate their request amounts towards the upstream (because there might not be any) and consumes the upstream in an unbounded manner (requesting Long.MAX_VALUE). Note that Subscribers receive a continuous sequence of values after they subscribed even if an individual item gets delayed due to backpressure. Due to concurrency requirements, a size-bounded ReplayProcessor may hold strong references to more source emissions than specified.

When this ReplayProcessor is terminated via onError(Throwable) or onComplete(), late Subscribers will receive the retained/cached items first (if any) followed by the respective terminal event. If the ReplayProcessor has a time-bound, the age of the retained/cached items are still considered when replaying and thus it may result in no items being emitted before the terminal event.

Once an Subscriber has subscribed, it will receive items continuously from that point on. Bounds only affect how many past items a new Subscriber will receive before it catches up with the live event feed.

Even though ReplayProcessor implements the Subscriber interface, calling onSubscribe is not required (Rule 2.12) if the processor is used as a standalone source. However, calling onSubscribe after the ReplayProcessor reached its terminal state will result in the given Subscription being canceled immediately.

Calling onNext(Object), onError(Throwable) and onComplete() is required to be serialized (called from the same thread or called non-overlappingly from different threads through external means of serialization). The toSerialized() method available to all FlowableProcessors provides such serialization and also protects against reentrance (i.e., when a downstream Subscriber consuming this processor also wants to call onNext(Object) on this processor recursively).

This ReplayProcessor supports the standard state-peeking methods hasComplete(), hasThrowable(), getThrowable() and hasSubscribers() as well as means to read the retained/cached items in a non-blocking and thread-safe manner via hasValue(), getValue(), getValues() or getValues(Object[]).

Note that due to concurrency requirements, a size- and time-bounded ReplayProcessor may hold strong references to more source emissions than specified while it isn't terminated yet. Use the cleanupBuffer() to allow such inaccessible items to be cleaned up by GC once no consumer references them anymore.

Backpressure:

This ReplayProcessor respects the individual backpressure behavior of its Subscribers but does not coordinate their request amounts towards the upstream (because there might not be any) and consumes the upstream in an unbounded manner (requesting Long.MAX_VALUE). Note that Subscribers receive a continuous sequence of values after they subscribed even if an individual item gets delayed due to backpressure.

Scheduler:

ReplayProcessor does not operate by default on a particular io.reactivex.Scheduler and the Subscribers get notified on the thread the respective onXXX methods were invoked. Time-bound ReplayProcessors use the given Scheduler in their create methods as time source to timestamp of items received for the age checks.

Error handling:

When the onError(Throwable) is called, the ReplayProcessor enters into a terminal state and emits the same Throwable instance to the last set of Subscribers. During this emission, if one or more Subscribers cancel their respective Subscriptions, the Throwable is delivered to the global error handler via io.reactivex.plugins.RxJavaPlugins#onError(Throwable) (multiple times if multiple Subscribers cancel at once). If there were no Subscribers subscribed to this ReplayProcessor when the onError() was called, the global error handler is not invoked.

Example usage:

 ReplayProcessor<Object> processor = new ReplayProcessor<T>();
 processor.onNext("one");
 processor.onNext("two");
 processor.onNext("three");
 processor.onComplete();

 // both of the following will get the onNext/onComplete calls from above
 processor.subscribe(subscriber1);
 processor.subscribe(subscriber2);