Package org.reactfx

Class SuspendableBase<O,T,A>

java.lang.Object

org.reactfx.ObservableBase<O,T>

org.reactfx.SuspendableBase<O,T,A>

All Implemented Interfaces:

Observable<O>, ProperObservable<O,T>, Suspendable

Direct Known Subclasses:

SuspendableEventStreamBase, SuspendableListWrapper, SuspendableValWrapper

public abstract class SuspendableBase<O,T,A>
extends ObservableBase<O,T>
implements Suspendable

Field Summary

Fields

Modifier and Type	Field	Description
private A	accumulatedValue
private AccumulationFacility<T,A>	af
private boolean	hasValue
private EventStream<T>	input
private int	suspended

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	SuspendableBase(EventStream<T> input, NotificationAccumulator<O,T,A> pn)

Method Summary

Modifier and Type	Method	Description
private void	handleEvent(T event)
protected abstract T	headOf(A accum)
protected A	initialAccumulator(T value)
protected boolean	isSuspended()
protected Subscription	observeInputs()	Starts observing this observable's input(s), if any.
protected A	reduce(A accum, T value)
private void	reset()
private void	resume()
protected abstract AccumulatorSize	sizeOf(A accum)
Guard	suspend()	Suspends this suspendable object.
protected abstract A	tailOf(A accum)

Methods inherited from class org.reactfx.ObservableBase

addObserver, enqueueNotifications, equals, forEachObserver, getObserverCount, hashCode, isObservingInputs, newObserver, notifyObservers, notifyObservers, observe, removeObserver, toString

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.reactfx.ProperObservable

defaultEquals, defaultHashCode, defaultNotificationAccumulator, defaultToString

Methods inherited from interface org.reactfx.Suspendable

suspendWhen, suspendWhile, suspendWhile

Field Detail

input

private final EventStream<T> input

af

private final AccumulationFacility<T,A> af

suspended

private int suspended

hasValue

private boolean hasValue

accumulatedValue

private A accumulatedValue

Constructor Detail

SuspendableBase

protected SuspendableBase(EventStream<T> input,
                          NotificationAccumulator<O,T,A> pn)

Method Detail

sizeOf

protected abstract AccumulatorSize sizeOf(A accum)

headOf

protected abstract T headOf(A accum)

tailOf

protected abstract A tailOf(A accum)

initialAccumulator

protected A initialAccumulator(T value)

reduce

protected A reduce(A accum,
                   T value)

isSuspended

protected final boolean isSuspended()

suspend

public final Guard suspend()

Description copied from interface: Suspendable

Suspends this suspendable object.

In case of suspendable Observable, suspends notification delivery for this observable object. Notifications produced while suspended may be queued for later delivery, accumulated into a single cumulative notification, or discarded completely, depending on the concrete implementation.

Specified by:

suspend in interface Suspendable

Returns:

a Guard instance that can be released to end suspension. In case of suspended notifications, releasing the returned Guard will trigger delivery of queued or accumulated notifications, if any.

The returned Guard is AutoCloseable, which makes it convenient to use in try-with-resources.

observeInputs

protected final Subscription observeInputs()

Description copied from class: ObservableBase

Starts observing this observable's input(s), if any. This method is called when the number of observers goes from 0 to 1. This method is called before ObservableBase.newObserver(Object) is called for the first observer.

Specified by:

observeInputs in class ObservableBase<O,T>

Returns:

subscription used to stop observing inputs. The subscription is unsubscribed (i.e. input observation stops) when the number of observers goes down to 0.

resume

private void resume()

reset

private void reset()

handleEvent

private void handleEvent(T event)