hu.akarnokd.reactive4java.util

Class DefaultConnectableObservable<T,U>

java.lang.Object

hu.akarnokd.reactive4java.util.DefaultConnectableObservable<T,U>

Type Parameters:

T - the type of values observed from the source

U - the type of values emmitted to registered observers

All Implemented Interfaces:

ConnectableObservable<U>, Observable<U>

public class DefaultConnectableObservable<T,U>
extends java.lang.Object
implements ConnectableObservable<U>

Default implementation of the connectable observable which can be disconnected from its source independently from the registered observers.

Since:

0.97

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	DefaultConnectableObservable.InnerConnection The inner connection that nulls out the parent class' connection and deregisters the subject, but only once.

Field Summary

Fields

Modifier and Type	Field and Description
protected java.io.Closeable	connection The active connection's close reference.
protected java.util.concurrent.locks.Lock	lock The lock.
protected Observable<? extends T>	source The source observable.
protected Subject<? super T,? extends U>	subject The subject that is connected to the source.

Constructor Summary

Constructors

Constructor and Description
DefaultConnectableObservable(Observable<? extends T> source, Subject<? super T,? extends U> subject) Creates an observable which can be connected and disconnected from the source.
DefaultConnectableObservable(Observable<? extends T> source, Subject<? super T,? extends U> subject, java.util.concurrent.locks.Lock lock) Creates an observable which can be connected and disconnected from the source.

Method Summary

Modifier and Type	Method and Description
java.io.Closeable	connect() Connects the wrapper to its source.
java.io.Closeable	register(Observer<? super U> observer) Registers an observer for the notification of Ts.

Methods inherited from class java.lang.Object

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

Field Detail

subject

@Nonnull
protected final Subject<? super T,? extends U> subject

The subject that is connected to the source.

source

@Nonnull
protected final Observable<? extends T> source

The source observable.

lock

@Nonnull
protected final java.util.concurrent.locks.Lock lock

The lock.

connection

@Nullable
protected java.io.Closeable connection

The active connection's close reference.

Constructor Detail

DefaultConnectableObservable

public DefaultConnectableObservable(@Nonnull
                                    Observable<? extends T> source,
                                    @Nonnull
                                    Subject<? super T,? extends U> subject)

Creates an observable which can be connected and disconnected from the source.

Uses fair ReentrantLock.

Parameters:

source - the underlying observable source of Ts

subject - the observer that receives values from the source in case it is connected

DefaultConnectableObservable

public DefaultConnectableObservable(@Nonnull
                                    Observable<? extends T> source,
                                    @Nonnull
                                    Subject<? super T,? extends U> subject,
                                    @Nonnull
                                    java.util.concurrent.locks.Lock lock)

Creates an observable which can be connected and disconnected from the source.

Parameters:

source - the underlying observable source of Ts

subject - the observer that receives values from the source in case it is connected

lock - the lock to use

Method Detail

connect

public java.io.Closeable connect()

Description copied from interface: ConnectableObservable

Connects the wrapper to its source. All registered observers will receive values while this connection is active. Closing the connection will stop the observers from receiving events, but they won't be deregistered. Connecting to an already connected source is a no-op.

Specified by:

connect in interface ConnectableObservable<U>

Returns:

the handler to close the connection

register

@Nonnull
public java.io.Closeable register(@Nonnull
                                            Observer<? super U> observer)

Description copied from interface: Observable

Registers an observer for the notification of Ts.

Specified by:

register in interface Observable<U>

Parameters:

observer - the observer of Ts or any supertype of it

Returns:

the way of deregister the observer from this provider