hu.akarnokd.reactive4java.util

Class DefaultObserverEx<T>

java.lang.Object

hu.akarnokd.reactive4java.util.DefaultObserver<T>

hu.akarnokd.reactive4java.util.DefaultObserverEx<T>

Type Parameters:

T - the element type

All Implemented Interfaces:

BaseObserver, Observer<T>, java.io.Closeable, java.lang.AutoCloseable

Direct Known Subclasses:

Zip.ManyObservables.ItemObserver

public abstract class DefaultObserverEx<T>
extends DefaultObserver<T>

Extension to the default observer for cases when the observer itself needs to track sub-observers and their close handlers.

Note when overriding onClose, you should always call super.onClose() to close any remaining subObservers.

Field Summary

Fields

Modifier and Type	Field and Description
protected TaggedCompositeCloseable	subObservers The container for the tagged observers.

Fields inherited from class hu.akarnokd.reactive4java.util.DefaultObserver

closeOnTermination, completed, lock

Constructor Summary

Constructors

Constructor and Description
DefaultObserverEx() Constructor.
DefaultObserverEx(boolean complete) Constructor.
DefaultObserverEx(java.util.concurrent.locks.Lock lock, boolean complete) Constructor.

Method Summary

Modifier and Type	Method and Description
void	add(java.lang.Object token, java.io.Closeable handler) Adds or replaces a new closeable with a token to the sub-observer list.
protected void	init() Implementations might override this method to initialize some other resources outside the onRegister's lock, before onRegister.
protected void	onClose() Called by close to close down any associated resources with this instance.
protected void	onRegister() Called internally with the global lock held to ensure any dependent registrations succeed to store the closeable reference before returning.
java.io.Closeable	registerWith(Observable<? extends T> source) While holding the global lock, executes the onRegister method then registers this instance with the supplied source observable with the this token.
void	remove(java.lang.Object token) Removes and closes the close handler associated with the token.
static <T> DefaultObserverEx<T>	wrap(Observer<? super T> observer) Wraps the supplied observer into a DefaultObservableEx and simply forwards onNext, onError, onFinish events to the raw next, error and finish methods.

Methods inherited from class hu.akarnokd.reactive4java.util.DefaultObserver

close, error, finish, getLock, isCompleted, next, onError, onFinish, onNext

Methods inherited from class java.lang.Object

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

Field Detail

subObservers

@Nonnull
protected final TaggedCompositeCloseable subObservers

The container for the tagged observers.

Constructor Detail

DefaultObserverEx

public DefaultObserverEx()

Constructor. Sets up the observer to complete once the error or finish messages are received.

DefaultObserverEx

public DefaultObserverEx(boolean complete)

Constructor.

Parameters:

complete - should the observer close its sub-resources automatically on error/finish?

DefaultObserverEx

public DefaultObserverEx(@Nonnull
                         java.util.concurrent.locks.Lock lock,
                         boolean complete)

Constructor.

Parameters:

lock - the external lock to use when synchronizing the message methods

complete - should the observer close its sub-resources automatically on error/finish?

Method Detail

add

public void add(java.lang.Object token,
                java.io.Closeable handler)

Adds or replaces a new closeable with a token to the sub-observer list. If this observer is already in completed state, the handler is closed immediately.

Parameters:

token - the reference token

handler - the closeable handler

onRegister

protected void onRegister()

Called internally with the global lock held to ensure any dependent registrations succeed to store the closeable reference before returning.

registerWith

@Nonnull
public java.io.Closeable registerWith(@Nonnull
                                                Observable<? extends T> source)

While holding the global lock, executes the onRegister method then registers this instance with the supplied source observable with the this token. If the this token is already registered, the previous handle is closed.

Parameters:

source - the source observable

Returns:

this

remove

public void remove(@Nonnull
                   java.lang.Object token)

Removes and closes the close handler associated with the token.

Parameters:

token - the token to the closeable handler

onClose

protected void onClose()

Description copied from class: DefaultObserver

Called by close to close down any associated resources with this instance.

The close() method ensures that the lock is held this code executes.

Overrides:

onClose in class DefaultObserver<T>

wrap

public static <T> DefaultObserverEx<T> wrap(@Nonnull
                                            Observer<? super T> observer)

Wraps the supplied observer into a DefaultObservableEx and simply forwards onNext, onError, onFinish events to the raw next, error and finish methods.

Type Parameters:

T - the observed element type

Parameters:

observer - the observer to wrap

Returns:

the wrapper default observer ex

init

protected void init()

Implementations might override this method to initialize some other resources outside the onRegister's lock, before onRegister.