Class AsyncSubscriber<T>

java.lang.Object

org.reactivestreams.example.unicast.AsyncSubscriber<T>

All Implemented Interfaces:

Runnable, Subscriber<T>

public abstract class AsyncSubscriber<T>
extends Object
implements Subscriber<T>, Runnable

AsyncSubscriber is an implementation of Reactive Streams `Subscriber`, it runs asynchronously (on an Executor), requests one element at a time, and invokes a user-defined method to process each element. NOTE: The code below uses a lot of try-catches to show the reader where exceptions can be expected, and where they are forbidden.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private static enum	AsyncSubscriber.OnComplete
private static class	AsyncSubscriber.OnError
private static class	AsyncSubscriber.OnNext<T>
private static class	AsyncSubscriber.OnSubscribe
private static interface	AsyncSubscriber.Signal

Field Summary

Fields

Modifier and Type	Field	Description
private boolean	done
private final Executor	executor
private final ConcurrentLinkedQueue<AsyncSubscriber.Signal>	inboundSignals
private final AtomicBoolean	on
private Subscription	subscription

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	AsyncSubscriber(Executor executor)

Method Summary

Modifier and Type	Method	Description
private final void	done()
private void	handleOnComplete()
private void	handleOnError(Throwable error)
private final void	handleOnNext(T element)
private final void	handleOnSubscribe(Subscription s)
final void	onComplete()	Successful terminal state.
final void	onError(Throwable t)	Failed terminal state.
final void	onNext(T element)	Data notification sent by the Publisher in response to requests to Subscription.request(long).
final void	onSubscribe(Subscription s)	Invoked after calling Publisher.subscribe(Subscriber).
final void	run()
private void	signal(AsyncSubscriber.Signal signal)
private final void	tryScheduleToExecute()
protected void	whenComplete()
protected void	whenError(Throwable error)
protected abstract boolean	whenNext(T element)

Methods inherited from class Object

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

Field Details

subscription

private Subscription subscription

done

private boolean done

executor

private final Executor executor

inboundSignals

private final ConcurrentLinkedQueue<AsyncSubscriber.Signal> inboundSignals

on

private final AtomicBoolean on

Constructor Details

AsyncSubscriber

protected AsyncSubscriber(Executor executor)

Method Details

done

private final void done()

whenNext

protected abstract boolean whenNext(T element)

whenComplete

protected void whenComplete()

whenError

protected void whenError(Throwable error)

handleOnSubscribe

private final void handleOnSubscribe(Subscription s)

handleOnNext

private final void handleOnNext(T element)

handleOnComplete

private void handleOnComplete()

handleOnError

private void handleOnError(Throwable error)

onSubscribe

public final void onSubscribe(Subscription s)

Description copied from interface: Subscriber

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).

Specified by:

onSubscribe in interface Subscriber<T>

Parameters:

s - the Subscription that allows requesting data via Subscription.request(long)

onNext

public final void onNext(T element)

Description copied from interface: Subscriber

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

Specified by:

onNext in interface Subscriber<T>

Parameters:

element - the element signaled

onError

public final void onError(Throwable t)

Description copied from interface: Subscriber

Failed terminal state.

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

Specified by:

onError in interface Subscriber<T>

Parameters:

t - the throwable signaled

onComplete

public final void onComplete()

Description copied from interface: Subscriber

Successful terminal state.

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

Specified by:

onComplete in interface Subscriber<T>

run

public final void run()

Specified by:

run in interface Runnable

signal

private void signal(AsyncSubscriber.Signal signal)

tryScheduleToExecute

private final void tryScheduleToExecute()