public final Class ReplaySubject<T>

This subject does not have a public constructor by design; a new empty instance of this ReplaySubject can be created via the following create methods that allow specifying the retention policy for items:

• create() - creates an empty, unbounded ReplaySubject that caches all items and the terminal event it receives.

  

  

• create(int) - creates an empty, unbounded ReplaySubject with a hint about how many total items one expects to retain.
• createWithSize(int) - creates an empty, size-bound ReplaySubject that retains at most the given number of the latest item it receives.

  

• createWithTime(long, TimeUnit, Scheduler) - creates an empty, time-bound ReplaySubject that retains items no older than the specified time amount.

  

• createWithTimeAndSize(long, TimeUnit, Scheduler, int) - creates an empty, time- and size-bound ReplaySubject that retains at most the given number items that are also not older than the specified time amount.

  

Since a Subject is conceptionally derived from the Processor type in the Reactive Streams specification, 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 subject's state is not changed.

Since a ReplaySubject is an io.reactivex.rxjava3.core.Observable, it does not support backpressure.

When this ReplaySubject is terminated via onError(Throwable) or onComplete(), late io.reactivex.rxjava3.core.Observers will receive the retained/cached items first (if any) followed by the respective terminal event. If the ReplaySubject 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 Observer has subscribed, it will receive items continuously from that point on. Bounds only affect how many past items a new Observer will receive before it catches up with the live event feed.

Even though ReplaySubject implements the Observer interface, calling onSubscribe is not required (Rule 2.12) if the subject is used as a standalone source. However, calling onSubscribe after the ReplaySubject reached its terminal state will result in the given Disposable being disposed 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 Subjects provides such serialization and also protects against reentrance (i.e., when a downstream Observer consuming this subject also wants to call onNext(Object) on this subject recursively).

This ReplaySubject supports the standard state-peeking methods hasComplete(), hasThrowable(), getThrowable() and hasObservers() 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 ReplaySubject 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 it anymore.

Scheduler:

ReplaySubject does not operate by default on a particular io.reactivex.rxjava3.core.Scheduler and the Observers get notified on the thread the respective onXXX methods were invoked. Time-bound ReplaySubjects 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 ReplaySubject enters into a terminal state and emits the same Throwable instance to the last set of Observers. During this emission, if one or more Observers dispose their respective Disposables, the Throwable is delivered to the global error handler via io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) (multiple times if multiple Observers cancel at once). If there were no Observers subscribed to this ReplaySubject when the onError() was called, the global error handler is not invoked.

Example usage:

 ReplaySubject<Object> subject = ReplaySubject.create();
 subject.onNext("one");
 subject.onNext("two");
 subject.onNext("three");
 subject.onComplete();

 // both of the following will get the onNext/onComplete calls from above
 subject.subscribe(observer1);
 subject.subscribe(observer2);