A Subject that queues up events until a single Observer subscribes to it, replays those events to it until the Observer catches up and then switches to relaying events live to this single Observer until this UnicastSubject terminates or the Observer unsubscribes.

Note that UnicastSubject holds an unbounded internal buffer.

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

• create() - creates an empty, unbounded UnicastSubject that caches all items and the terminal event it receives.
• create(int) - creates an empty, unbounded UnicastSubject with a hint about how many total items one expects to retain.
• create(boolean) - creates an empty, unbounded UnicastSubject that optionally delays an error it receives and replays it after the regular items have been emitted.
• create(int, Runnable) - creates an empty, unbounded UnicastSubject with a hint about how many total items one expects to retain and a callback that will be called exactly once when the UnicastSubject gets terminated or the single Observer unsubscribes.
• create(int, Runnable, boolean) - creates an empty, unbounded UnicastSubject with a hint about how many total items one expects to retain and a callback that will be called exactly once when the UnicastSubject gets terminated or the single Observer unsubscribes and optionally delays an error it receives and replays it after the regular items have been emitted.

If more than one Observer attempts to subscribe to this UnicastSubject, they will receive an IllegalStateException indicating the single-use-only nature of this UnicastSubject, even if the UnicastSubject already terminated with an error.

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 UnicastSubject is an io.reactivex.Observable, it does not support backpressure.

When this UnicastSubject is terminated via onError(Throwable) the current or late single Observer may receive the Throwable before any available items could be emitted. To make sure an onError event is delivered to the Observer after the normal items, create a UnicastSubject with the create(boolean) or create(int, Runnable, boolean) factory methods.

Even though UnicastSubject 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 UnicastSubject 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 UnicastSubject supports the standard state-peeking methods hasComplete(), hasThrowable(), getThrowable() and hasObservers().

Scheduler:

UnicastSubject does not operate by default on a particular io.reactivex.Scheduler and the single Observer gets notified on the thread the respective onXXX methods were invoked.

Error handling:

When the onError(Throwable) is called, the UnicastSubject enters into a terminal state and emits the same Throwable instance to the current single Observer. During this emission, if the single Observers disposes its respective Disposable, the Throwable is delivered to the global error handler via io.reactivex.plugins.RxJavaPlugins#onError(Throwable). If there were no Observers subscribed to this UnicastSubject when the onError() was called, the global error handler is not invoked.

Example usage:

UnicastSubject<Integer> subject = UnicastSubject.create();

TestObserver<Integer> to1 = subject.test();

// fresh UnicastSubjects are empty
to1.assertEmpty();

TestObserver<Integer> to2 = subject.test();

// A UnicastSubject only allows one Observer during its lifetime
to2.assertFailure(IllegalStateException.class);

subject.onNext(1);
to1.assertValue(1);

subject.onNext(2);
to1.assertValues(1, 2);

subject.onComplete();
to1.assertResult(1, 2);

// ----------------------------------------------------

UnicastSubject<Integer> subject2 = UnicastSubject.create();

// a UnicastSubject caches events until its single Observer subscribes
subject2.onNext(1);
subject2.onNext(2);
subject2.onComplete();

TestObserver<Integer> to3 = subject2.test();

// the cached events are emitted in order
to3.assertResult(1, 2);