reactor.core.publisher

Class BaseSubscriber<T>

java.lang.Object

reactor.core.publisher.BaseSubscriber<T>

All Implemented Interfaces:

Subscriber<T>, Subscription, CoreSubscriber<T>, Disposable

public abstract class BaseSubscriber<T>
extends Object
implements CoreSubscriber<T>, Subscription, Disposable

A simple base class for a Subscriber implementation that lets the user perform a request(long) and cancel() on it directly. As the targeted use case is to manually handle requests, the hookOnSubscribe(Subscription) and hookOnNext(Object) hooks are expected to be implemented, but they nonetheless default to an unbounded request at subscription time. If you need to define a Context for this BaseSubscriber, simply override its CoreSubscriber.currentContext() method.

Override the other optional hooks hookOnComplete(), hookOnError(Throwable) and hookOnCancel() to customize the base behavior. You also have a termination hook, hookFinally(SignalType).

Most of the time, exceptions triggered inside hooks are propagated to onError(Throwable) (unless there is a fatal exception). The class is in the reactor.core.publisher package, as this subscriber is tied to a single Publisher.

Author:

Simon Baslé

Nested Class Summary

Nested classes/interfaces inherited from interface reactor.core.Disposable

Disposable.Composite, Disposable.Swap

Constructor Summary

Constructors

Constructor and Description
BaseSubscriber()

Method Summary

Modifier and Type	Method and Description
void	cancel()
void	dispose() Dispose the Subscription by cancelling it.
protected void	hookFinally(SignalType type) Optional hook executed after any of the termination events (onError, onComplete, cancel).
protected void	hookOnCancel() Optional hook executed when the subscription is cancelled by calling this Subscriber's cancel() method.
protected void	hookOnComplete() Optional hook for completion processing.
protected void	hookOnError(Throwable throwable) Optional hook for error processing.
protected void	hookOnNext(T value) Hook for processing of onNext values.
protected void	hookOnSubscribe(Subscription subscription) Hook for further processing of onSubscribe's Subscription.
boolean	isDisposed() Optionally return true when the resource or task is disposed.
void	onComplete()
void	onError(Throwable t)
void	onNext(T value)
void	onSubscribe(Subscription s) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
void	request(long n)
void	requestUnbounded() Request an unbounded amount.
String	toString()
protected Subscription	upstream() Return current Subscription

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface reactor.core.CoreSubscriber

currentContext

Constructor Detail

BaseSubscriber

public BaseSubscriber()

Method Detail

upstream

protected Subscription upstream()

Return current Subscription

Returns:

current Subscription

isDisposed

public boolean isDisposed()

Description copied from interface: Disposable

Optionally return true when the resource or task is disposed.

Implementations are not required to track disposition and as such may never return true even when disposed. However, they MUST only return true when there's a guarantee the resource or task is disposed.

Specified by:

isDisposed in interface Disposable

Returns:

true when there's a guarantee the resource or task is disposed.

dispose

public void dispose()

Dispose the Subscription by cancelling it.

Specified by:

dispose in interface Disposable

hookOnSubscribe

protected void hookOnSubscribe(Subscription subscription)

Hook for further processing of onSubscribe's Subscription. Implement this method to call request(long) as an initial request. Values other than the unbounded Long.MAX_VALUE imply that you'll also call request in hookOnNext(Object).

Defaults to request unbounded Long.MAX_VALUE as in requestUnbounded()

Parameters:

subscription - the subscription to optionally process

hookOnNext

protected void hookOnNext(T value)

Hook for processing of onNext values. You can call request(long) here to further request data from the source Publisher if the initial request wasn't unbounded.

Defaults to doing nothing.

Parameters:

value - the emitted value to process

hookOnComplete

protected void hookOnComplete()

Optional hook for completion processing. Defaults to doing nothing.

hookOnError

protected void hookOnError(Throwable throwable)

Optional hook for error processing. Default is to call Exceptions.errorCallbackNotImplemented(Throwable).

Parameters:

throwable - the error to process

hookOnCancel

protected void hookOnCancel()

Optional hook executed when the subscription is cancelled by calling this Subscriber's cancel() method. Defaults to doing nothing.

hookFinally

protected void hookFinally(SignalType type)

Optional hook executed after any of the termination events (onError, onComplete, cancel). The hook is executed in addition to and after hookOnError(Throwable), hookOnComplete() and hookOnCancel() hooks, even if these callbacks fail. Defaults to doing nothing. A failure of the callback will be caught by Operators.onErrorDropped(Throwable, reactor.util.context.Context).

Parameters:

type - the type of termination event that triggered the hook (SignalType.ON_ERROR, SignalType.ON_COMPLETE or SignalType.CANCEL)

onSubscribe

public final void onSubscribe(Subscription s)

Description copied from interface: CoreSubscriber

Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long). Should further onNext related state modification occur, thread-safety will be required.

Note that an invalid request <= 0 will not produce an onError and will simply be ignored or reported through a debug-enabled Logger.

Specified by:

onSubscribe in interface Subscriber<T>

Specified by:

onSubscribe in interface CoreSubscriber<T>

onNext

public final void onNext(T value)

Specified by:

onNext in interface Subscriber<T>

onError

public final void onError(Throwable t)

Specified by:

onError in interface Subscriber<T>

onComplete

public final void onComplete()

Specified by:

onComplete in interface Subscriber<T>

request

public final void request(long n)

Specified by:

request in interface Subscription

requestUnbounded

public final void requestUnbounded()

Request an unbounded amount.

cancel

public final void cancel()

Specified by:

cancel in interface Subscription

toString

public String toString()

Overrides:

toString in class Object