Top Description Inners Fields Constructors Methods
rx.subjects

pack-priv final Class SubjectSubscriptionManager<T>

extends AtomicReference<SubjectSubscriptionManager.State<T>>
implements Observable.OnSubscribe<T>
Class Inheritance
All Implemented Interfaces
rx.Observable.OnSubscribe, rx.functions.Action1, rx.functions.Action, rx.functions.Function
Annotations
@SuppressWarnings:unchecked, rawtypes
Type Parameters
<T>
the source and return value type
Imports
java.util.ArrayList, .List, java.util.concurrent.atomic.AtomicReference, rx.Observable.OnSubscribe, .Observer, .Subscriber, rx.functions.Action0, .Action1, .Actions, rx.internal.operators.NotificationLite, rx.subscriptions.Subscriptions
Represents the typical state and OnSubscribe logic for a Subject implementation.

Nested and Inner Type Summary

Modifier and TypeClass and Description
protected static class
SubjectSubscriptionManager.State<T>

State-machine representing the termination state and active SubjectObservers.

protected static class
SubjectSubscriptionManager.SubjectObserver<
the consumed value type of the actual Observer
T>

Observer wrapping the actual Subscriber and providing various emission facilities.

Field Summary

Modifier and TypeField and Description
pack-priv boolean
active

Indicates that the subject is active (cheaper than checking the state).

pack-priv volatile Object
latest

Stores the latest value or the terminal value for some Subjects.

pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>>
onAdded

Action called after the subscriber has been added to the state.

pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>>
onStart

Action called when a new subscriber subscribes but before it is added to the state.

pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>>
onTerminated

Action called when the subscriber wants to subscribe to a terminal state.

private static final long
serialVersionUID

Hides java.util.concurrent.atomic.AtomicReference.serialVersionUID.

Constructor Summary

AccessConstructor and Description
public
SubjectSubscriptionManager()

Method Summary

Modifier and TypeMethod and Description
pack-priv boolean

Returns:

false if the subject is already in its terminal state
add(SubjectSubscriptionManager.SubjectObserver<T>
the SubjectObserver to add
o)

Try to atomically add a SubjectObserver to the active state.

pack-priv void
addUnsubscriber(Subscriber<? super T> child, final SubjectSubscriptionManager.SubjectObserver<T> bo)

Registers the unsubscribe action for the given subscriber.

public void
call(final Subscriber<? super T> child)

Implements rx.functions.Action1.call.

pack-priv Object

Returns:

Retrieve the latest NotificationLite value
getLatest()

pack-priv SubjectSubscriptionManager.SubjectObserver<T>[]

Returns:

the array of SubjectObservers, don't write into the array!
next(Object
the new latest value
n)

Set a new latest NotificationLite value and return the active observers.

pack-priv SubjectSubscriptionManager.SubjectObserver<T>[]

Returns:

the array of active subscribers, don't write into the array!
observers()

pack-priv void
remove(SubjectSubscriptionManager.SubjectObserver<T>
the SubjectObserver to remove
o)

Atomically remove the specified SubjectObserver from the active observers.

pack-priv void
setLatest(Object value)

Set the latest NotificationLite value.

pack-priv SubjectSubscriptionManager.SubjectObserver<T>[]

Returns:

the last active SubjectObservers
terminate(Object
the terminal value
n)

Atomically set the terminal NotificationLite value (which could be any of the 3), clear the active observers and return the last active observers.

Inherited from java.util.concurrent.atomic.AtomicReference:
accumulateAndGetcompareAndExchangecompareAndExchangeAcquirecompareAndExchangeReleasecompareAndSetgetgetAcquiregetAndAccumulategetAndSetgetAndUpdategetOpaquegetPlainlazySetsetsetOpaquesetPlainsetReleasetoStringupdateAndGetweakCompareAndSetweakCompareAndSetAcquireweakCompareAndSetPlainweakCompareAndSetReleaseweakCompareAndSetVolatile

Field Detail

activeback to summary
pack-priv boolean active

Indicates that the subject is active (cheaper than checking the state).

latestback to summary
pack-priv volatile Object latest

Stores the latest value or the terminal value for some Subjects.

onAddedback to summary
pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>> onAdded

Action called after the subscriber has been added to the state.

onStartback to summary
pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>> onStart

Action called when a new subscriber subscribes but before it is added to the state.

onTerminatedback to summary
pack-priv Action1<SubjectSubscriptionManager.SubjectObserver<T>> onTerminated

Action called when the subscriber wants to subscribe to a terminal state.

serialVersionUIDback to summary
private static final long serialVersionUID

Hides java.util.concurrent.atomic.AtomicReference.serialVersionUID.

Constructor Detail

SubjectSubscriptionManagerback to summary
public SubjectSubscriptionManager()

Method Detail

addback to summary
pack-priv boolean add(SubjectSubscriptionManager.SubjectObserver<T> o)

Try to atomically add a SubjectObserver to the active state.

Parameters
o:SubjectSubscriptionManager.SubjectObserver<T>

the SubjectObserver to add

Returns:boolean

false if the subject is already in its terminal state

addUnsubscriberback to summary
pack-priv void addUnsubscriber(Subscriber<? super T> child, final SubjectSubscriptionManager.SubjectObserver<T> bo)

Registers the unsubscribe action for the given subscriber.

callback to summary
public void call(final Subscriber<? super T> child)

Implements rx.functions.Action1.call.

Annotations
@Override
getLatestback to summary
pack-priv Object getLatest()
Returns:Object

Retrieve the latest NotificationLite value

nextback to summary
pack-priv SubjectSubscriptionManager.SubjectObserver<T>[] next(Object n)

Set a new latest NotificationLite value and return the active observers.

Parameters
n:Object

the new latest value

Returns:SubjectSubscriptionManager.SubjectObserver<T>[]

the array of SubjectObservers, don't write into the array!

observersback to summary
pack-priv SubjectSubscriptionManager.SubjectObserver<T>[] observers()
Returns:SubjectSubscriptionManager.SubjectObserver<T>[]

the array of active subscribers, don't write into the array!

removeback to summary
pack-priv void remove(SubjectSubscriptionManager.SubjectObserver<T> o)

Atomically remove the specified SubjectObserver from the active observers.

Parameters
o:SubjectSubscriptionManager.SubjectObserver<T>

the SubjectObserver to remove

setLatestback to summary
pack-priv void setLatest(Object value)

Set the latest NotificationLite value.

terminateback to summary
pack-priv SubjectSubscriptionManager.SubjectObserver<T>[] terminate(Object n)

Atomically set the terminal NotificationLite value (which could be any of the 3), clear the active observers and return the last active observers.

Parameters
n:Object

the terminal value

Returns:SubjectSubscriptionManager.SubjectObserver<T>[]

the last active SubjectObservers

rx.subjects back to summary

protected final Class SubjectSubscriptionManager.State<T>

extends Object
Class Inheritance
State-machine representing the termination state and active SubjectObservers.

Field Summary

Modifier and TypeField and Description
pack-priv static final SubjectSubscriptionManager.State<T>
EMPTY

pack-priv static final SubjectSubscriptionManager.SubjectObserver<T>[]
NO_OBSERVERS

pack-priv final SubjectSubscriptionManager.SubjectObserver<T>[]
observers

pack-priv final boolean
terminated

pack-priv static final SubjectSubscriptionManager.State<T>
TERMINATED

Constructor Summary

AccessConstructor and Description
public
State(boolean terminated, SubjectSubscriptionManager.SubjectObserver<T>[] observers)

Method Summary

Modifier and TypeMethod and Description
public SubjectSubscriptionManager.State<T>
add(SubjectSubscriptionManager.SubjectObserver<T> o)

public SubjectSubscriptionManager.State<T>
remove(SubjectSubscriptionManager.SubjectObserver<T> o)

Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

EMPTYback to summary
pack-priv static final SubjectSubscriptionManager.State<T> EMPTY
NO_OBSERVERSback to summary
pack-priv static final SubjectSubscriptionManager.SubjectObserver<T>[] NO_OBSERVERS
observersback to summary
pack-priv final SubjectSubscriptionManager.SubjectObserver<T>[] observers
terminatedback to summary
pack-priv final boolean terminated
TERMINATEDback to summary
pack-priv static final SubjectSubscriptionManager.State<T> TERMINATED

Constructor Detail

Stateback to summary
public State(boolean terminated, SubjectSubscriptionManager.SubjectObserver<T>[] observers)

Method Detail

addback to summary
public SubjectSubscriptionManager.State<T> add(SubjectSubscriptionManager.SubjectObserver<T> o)
removeback to summary
public SubjectSubscriptionManager.State<T> remove(SubjectSubscriptionManager.SubjectObserver<T> o)
rx.subjects back to summary

protected final Class SubjectSubscriptionManager.SubjectObserver<T>

extends Object
implements Observer<T>
Class Inheritance
All Implemented Interfaces
rx.Observer
Type Parameters
<T>
the consumed value type of the actual Observer
Observer wrapping the actual Subscriber and providing various emission facilities.

Field Summary

Modifier and TypeField and Description
pack-priv final Subscriber<? super T>
actual

The actual Observer.

pack-priv volatile boolean
caughtUp

Indicate that the observer has caught up.

pack-priv boolean
emitting

Guarded by this.

pack-priv boolean
fastPath

pack-priv boolean
first

Was the emitFirst run?

private volatile Object
index

Indicate where the observer is at replaying.

pack-priv List<Object>
queue

Guarded by this.

Constructor Summary

AccessConstructor and Description
public
SubjectObserver(Subscriber<? super T> actual)

Method Summary

Modifier and TypeMethod and Description
pack-priv void
accept(Object
the value to dispatch
n)

Dispatches a NotificationLite value to the actual Observer.

pack-priv void
emitFirst(Object
the type-appropriate notification lite object
n)

Tries to emit a NotificationLite value as the first value and drains the queue as long as possible.

pack-priv void
emitLoop(List<Object>
the initial queue contents
localQueue, Object
the current content to emit
current)

Emits the contents of the queue as long as there are values.

pack-priv void
emitNext(Object
the NotificationLite value
n)

Emits the given NotificationLite value and prevents the emitFirst to run if not already run.

pack-priv Observer<? super T>

Returns:

the actual Observer.
getActual()

public <
the index type
I> I

Returns:

the index value
index()

Returns the stored index.

public void
index(Object
the new index value
newIndex)

Sets a new index value.

public void
onCompleted()

Implements rx.Observer.onCompleted.

Notifies the Observer that the Observable has finished sending push-based notifications.
public void
onError(Throwable
the exception encountered by the Observable
e)

Implements rx.Observer.onError.

Notifies the Observer that the Observable has experienced an error condition.
public void
onNext(T
the item emitted by the Observable
t)

Implements rx.Observer.onNext.

Provides the Observer with a new item to observe.
Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

actualback to summary
pack-priv final Subscriber<? super T> actual

The actual Observer.

caughtUpback to summary
pack-priv volatile boolean caughtUp

Indicate that the observer has caught up.

emittingback to summary
pack-priv boolean emitting

Guarded by this.

fastPathback to summary
pack-priv boolean fastPath
firstback to summary
pack-priv boolean first

Was the emitFirst run? Guarded by this.

indexback to summary
private volatile Object index

Indicate where the observer is at replaying.

queueback to summary
pack-priv List<Object> queue

Guarded by this.

Constructor Detail

SubjectObserverback to summary
public SubjectObserver(Subscriber<? super T> actual)

Method Detail

acceptback to summary
pack-priv void accept(Object n)

Dispatches a NotificationLite value to the actual Observer.

Parameters
n:Object

the value to dispatch

emitFirstback to summary
pack-priv void emitFirst(Object n)

Tries to emit a NotificationLite value as the first value and drains the queue as long as possible.

Parameters
n:Object

the type-appropriate notification lite object

emitLoopback to summary
pack-priv void emitLoop(List<Object> localQueue, Object current)

Emits the contents of the queue as long as there are values.

Parameters
localQueue:List<Object>

the initial queue contents

current:Object

the current content to emit

emitNextback to summary
pack-priv void emitNext(Object n)

Emits the given NotificationLite value and prevents the emitFirst to run if not already run.

Parameters
n:Object

the NotificationLite value

getActualback to summary
pack-priv Observer<? super T> getActual()
Returns:Observer<? super T>

the actual Observer.

indexback to summary
public <I> I index()

Returns the stored index.

Parameters
<I>
the index type
Returns:I

the index value

indexback to summary
public void index(Object newIndex)

Sets a new index value.

Parameters
newIndex:Object

the new index value

onCompletedback to summary
public void onCompleted()

Implements rx.Observer.onCompleted.

Doc from rx.Observer.onCompleted.

Notifies the Observer that the Observable has finished sending push-based notifications.

The Observable will not call this method if it calls onError.

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

Implements rx.Observer.onError.

Doc from rx.Observer.onError.

Notifies the Observer that the Observable has experienced an error condition.

If the Observable calls this method, it will not thereafter call onNext or onCompleted.

Parameters
e:Throwable

the exception encountered by the Observable

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

Implements rx.Observer.onNext.

Doc from rx.Observer.onNext.

Provides the Observer with a new item to observe.

The Observable may call this method 0 or more times.

The Observable will not call this method again after it calls either onCompleted or onError.

Parameters
t:T

the item emitted by the Observable

Annotations
@Override