Abstract base implementation of an Observer with support for cancelling a subscription via cancel() (synchronously) and calls onStart() when the subscription happens.

All pre-implemented final methods are thread-safe.

Use the protected cancel() to dispose the sequence from within an onNext implementation.

Like all other consumers, DefaultObserver can be subscribed only once. Any subsequent attempt to subscribe it to a new source will yield an IllegalStateException with message "It is not allowed to subscribe with a(n) <class name> multiple times.".

Implementation of onStart(), onNext(Object), onError(Throwable) and onComplete() are not allowed to throw any unchecked exceptions. If for some reason this can't be avoided, use io.reactivex.rxjava3.core.Observable#safeSubscribe(io.reactivex.rxjava3.core.Observer) instead of the standard subscribe() method.

Example

Observable.range(1, 5)
    .subscribe(new DefaultObserver<Integer>() {
        @Override public void onStart() {
            System.out.println("Start!");
        }
        @Override public void onNext(Integer t) {
            if (t == 3) {
                cancel();
            }
            System.out.println(t);
        }
        @Override public void onError(Throwable t) {
            t.printStackTrace();
        }
        @Override public void onComplete() {
            System.out.println("Done!");
        }
    });