Top Description Fields Constructors Methods
reactor.core.publisher

public abstract Class BaseSubscriber<T>

extends Object
implements CoreSubscriber<T>, Subscription, Disposable
Class Inheritance
All Implemented Interfaces
reactor.core.Disposable, org.reactivestreams.Subscription, reactor.core.CoreSubscriber, org.reactivestreams.Subscriber
Imports
java.util.Objects, java.util.concurrent.atomic.AtomicReferenceFieldUpdater, org.reactivestreams.Subscriber, .Subscription, reactor.core.CoreSubscriber, .Disposable, .Exceptions, reactor.util.context.Context
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 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 org.reactivestreams.Publisher.

Author
Simon Baslé

Field Summary

Modifier and TypeField and Description
pack-priv static AtomicReferenceFieldUpdater<BaseSubscriber<T>, Subscription>
S

pack-priv volatile Subscription
subscription

Constructor Summary

AccessConstructor and Description
public
BaseSubscriber()

Method Summary

Modifier and TypeMethod and Description
public final void
cancel()

Implements org.reactivestreams.Subscription.cancel.

Request the Publisher to stop sending data and clean up resources.
public void
dispose()

Implements reactor.core.Disposable.dispose.

Dispose the Subscription by cancelling it.
protected void
hookFinally(SignalType
the type of termination event that triggered the hook (SignalType#ON_ERROR, SignalType#ON_COMPLETE or SignalType#CANCEL)
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
the error to process
throwable)

Optional hook for error processing.

protected void
hookOnNext(T
the emitted value to process
value)

Hook for processing of onNext values.

protected void
hookOnSubscribe(Subscription
the subscription to optionally process
subscription)

Hook for further processing of onSubscribe's Subscription.

public boolean
isDisposed()

Overrides default reactor.core.Disposable.isDisposed.

Optionally return true when the resource or task is disposed.
public final void
onComplete()

Implements org.reactivestreams.Subscriber.onComplete.

Successful terminal state.
public final void
onError(Throwable
the throwable signaled
t)

Implements org.reactivestreams.Subscriber.onError.

Failed terminal state.
public final void
onNext(T
the element signaled
value)

Implements org.reactivestreams.Subscriber.onNext.

Data notification sent by the Publisher in response to requests to Subscription#request(long).
public final void
onSubscribe(Subscription
the Subscription that allows requesting data via Subscription#request(long)
s)

Implements reactor.core.CoreSubscriber.onSubscribe.

Implementors should initialize any state used by onNext(Object) before calling Subscription#request(long).
public final void
request(long
the strictly positive number of elements to requests to the upstream Publisher
n)

Implements org.reactivestreams.Subscription.request.

No events will be sent by a Publisher until demand is signaled via this method.
public final void
requestUnbounded()

Request an unbounded amount.

pack-priv void
safeHookFinally(SignalType type)

public String
toString()

Overrides java.lang.Object.toString.

Returns a string representation of the object.
protected Subscription

Returns:

current Subscription
upstream()

Return current Subscription

Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAllwaitwaitwait

Field Detail

Sback to summary
pack-priv static AtomicReferenceFieldUpdater<BaseSubscriber<T>, Subscription> S
subscriptionback to summary
pack-priv volatile Subscription subscription

Constructor Detail

BaseSubscriberback to summary
public BaseSubscriber()

Method Detail

cancelback to summary
public final void cancel()

Implements org.reactivestreams.Subscription.cancel.

Doc from org.reactivestreams.Subscription.cancel.

Request the Publisher to stop sending data and clean up resources.

Data may still be sent to meet previously signalled demand after calling cancel.

Annotations
@Override
disposeback to summary
public void dispose()

Implements reactor.core.Disposable.dispose.

Dispose the Subscription by cancelling it.

Annotations
@Override
hookFinallyback to summary
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:SignalType

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

hookOnCancelback to summary
protected void hookOnCancel()

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

hookOnCompleteback to summary
protected void hookOnComplete()

Optional hook for completion processing. Defaults to doing nothing.

hookOnErrorback to summary
protected void hookOnError(Throwable throwable)

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

Parameters
throwable:Throwable

the error to process

hookOnNextback to summary
protected void hookOnNext(T value)

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

Defaults to doing nothing.

Parameters
value:T

the emitted value to process

hookOnSubscribeback to summary
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:Subscription

the subscription to optionally process

isDisposedback to summary
public boolean isDisposed()

Overrides default reactor.core.Disposable.isDisposed.

Doc from reactor.core.Disposable.isDisposed.

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.

Returns:boolean

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

Annotations
@Override
onCompleteback to summary
public final void onComplete()

Implements org.reactivestreams.Subscriber.onComplete.

Doc from org.reactivestreams.Subscriber.onComplete.

Successful terminal state.

No further events will be sent even if Subscription#request(long) is invoked again.

Annotations
@Override
onErrorback to summary
public final void onError(Throwable t)

Implements org.reactivestreams.Subscriber.onError.

Doc from org.reactivestreams.Subscriber.onError.

Failed terminal state.

No further events will be sent even if Subscription#request(long) is invoked again.

Parameters
t:Throwable

the throwable signaled

Annotations
@Override
onNextback to summary
public final void onNext(T value)

Implements org.reactivestreams.Subscriber.onNext.

Doc from org.reactivestreams.Subscriber.onNext.

Data notification sent by the Publisher in response to requests to Subscription#request(long).

Parameters
value:T

the element signaled

Annotations
@Override
onSubscribeback to summary
public final void onSubscribe(Subscription s)

Implements reactor.core.CoreSubscriber.onSubscribe.

Doc from reactor.core.CoreSubscriber.onSubscribe.

Implementors should initialize any state used by 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 reactor.util.Logger. Invoked after calling Publisher#subscribe(Subscriber).

No data will start flowing until Subscription#request(long) is invoked.

It is the responsibility of this Subscriber instance to call Subscription#request(long) whenever more data is wanted.

The Publisher will send notifications only in response to Subscription#request(long).

Parameters
s:Subscription

Doc from org.reactivestreams.Subscriber.onSubscribe.

the Subscription that allows requesting data via Subscription#request(long)

Annotations
@Override
requestback to summary
public final void request(long n)

Implements org.reactivestreams.Subscription.request.

Doc from org.reactivestreams.Subscription.request.

No events will be sent by a Publisher until demand is signaled via this method.

It can be called however often and whenever needed—but if the outstanding cumulative demand ever becomes Long.MAX_VALUE or more, it may be treated by the Publisher as "effectively unbounded".

Whatever has been requested can be sent by the Publisher so only signal demand for what can be safely handled.

A Publisher can send less than is requested if the stream ends but then must emit either Subscriber#onError(Throwable) or Subscriber#onComplete().

Parameters
n:long

the strictly positive number of elements to requests to the upstream Publisher

Annotations
@Override
requestUnboundedback to summary
public final void requestUnbounded()

Request an unbounded amount.

safeHookFinallyback to summary
pack-priv void safeHookFinally(SignalType type)
toStringback to summary
public String toString()

Overrides java.lang.Object.toString.

Doc from java.lang.Object.toString.

Returns a string representation of the object.

Returns:String

a string representation of the object.

Annotations
@Override
upstreamback to summary
protected Subscription upstream()

Return current Subscription

Returns:Subscription

current Subscription