hu.akarnokd.reactive4java.util

Class Observers

java.lang.Object

hu.akarnokd.reactive4java.util.Observers

public final class Observers
extends java.lang.Object

Utility class that helps in creating various observer instances.

Since:

0.97

Method Summary

Modifier and Type	Method and Description
static <T> Observer<T>	newAsyncAwaiter(java.util.concurrent.CountDownLatch latch) Creates an observer which counts down the latch by one in case of an error() or finish() event.
static <T> Observer<T>	newAsyncAwaiter(java.util.concurrent.locks.Lock lock, java.util.concurrent.locks.Condition cond) Creates an observer which signals the condition in case of an error() or finish() event.
static <T> Observer<T>	newAsyncAwaiter(java.lang.Object syncObject) Creates an observer which calls notifyAll on the supplied synchronization object in case of an error() or finish() event.
static <T> Observer<T>	newAsyncAwaiter(java.lang.Runnable onDone) Creates an observer which calls the supplied callback in case of an error() or finish() event.
static <T> Observer<T>	newAsyncAwaiter0(Action0 onDone) Creates an observer which calls the supplied callback in case of an error() or finish() event.
static <T> Observer<T>	print() Creates an observer with debugging purposes.
static <T> Observer<T>	print(java.lang.String separator, int maxLineLength) Creates an observer with debugging purposes.
static <T> Observer<T>	println() Creates an observer with debugging purposes.
static <T> Observer<T>	println(java.lang.String prefix) Creates an observer with debugging purposes.
static <T> java.io.Closeable	registerSafe(Observable<? extends T> source, Observer<? super T> observer) Registers the observer with the source in a way that if the registration call throws an exception, it is routed to the observer via the error() method.
static <T> java.io.Closeable	registerWith(java.lang.Iterable<T> iterable, Observer<? super T> observer) "Registers" the observer with the iterable source and iterates over the source iterable in the current thread.
static <T> java.io.Closeable	registerWith(java.lang.Iterable<T> iterable, Observer<? super T> observer, Scheduler scheduler) "Registers" an observer with the given iterable and iterates over it on the given scheduler.
static <T> java.io.Closeable	registerWith(Observable<T> observable, java.util.Observer javaObserver) Convenience method to register a java-observable with a reactive-observable.
static java.io.Closeable	registerWith(java.util.Observable javaObservable, java.util.Observer javaObserver) Convenience method to register a java-observer with a java-observable in the same style as the reactive counterpart do.
static <T> java.io.Closeable	registerWith(java.util.Observable javaObservable, Observer<T> observer) Convenience method to let a reactive-observer register with a java-observable.
static <T,U> Observer<U>	select(Observer<? super T> observer, Func1<? super U,? extends T> selector) Modifies the source observer to use the selector to compute a new value from the received one.
static <T> java.util.Observer	toJavaObserver(Observer<T> observer) Wraps the reactive-observer into a java-observer.
static <T> Observer<T>	toObserver(Action1<? super T> action) Wraps the given action as an observable which reacts only to next() events.
static <T> Observer<T>	toObserver(Action1<? super T> next, Action1<? super java.lang.Throwable> error, Action0 finish) Creates an observer which calls the given functions on its similarly named methods.
static OriginalObserverWrapper	toObserver(java.util.Observer javaObserver, java.util.Observable javaObservable) Converts the original Java Observer into a reactive-observer.

Methods inherited from class java.lang.Object

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

Method Detail

toObserver

@Nonnull
public static <T> Observer<T> toObserver(@Nonnull
                                                   Action1<? super T> action)

Wraps the given action as an observable which reacts only to next() events.

Type Parameters:

T - the type of the values

Parameters:

action - the action to wrap

Returns:

the observer wrapping the action

toObserver

@Nonnull
public static <T> Observer<T> toObserver(@Nonnull
                                                   Action1<? super T> next,
                                                   @Nonnull
                                                   Action1<? super java.lang.Throwable> error,
                                                   @Nonnull
                                                   Action0 finish)

Creates an observer which calls the given functions on its similarly named methods.

Type Parameters:

T - the value type to receive

Parameters:

next - the action to invoke on next()

error - the action to invoke on error()

finish - the action to invoke on finish()

Returns:

the observer

print

@Nonnull
public static <T> Observer<T> print()

Creates an observer with debugging purposes. It prints the submitted values to STDOUT separated by commas and line-broken by 80 characters, the exceptions to STDERR and prints an empty newline when it receives a finish().

Type Parameters:

T - the value type

Returns:

the observer

print

@Nonnull
public static <T> Observer<T> print(java.lang.String separator,
                                              int maxLineLength)

Creates an observer with debugging purposes. It prints the submitted values to STDOUT, the exceptions to STDERR and prints an empty newline when it receives a finish().

Type Parameters:

T - the value type

Parameters:

separator - the separator to use between subsequent values

maxLineLength - how many characters to print into each line

Returns:

the observer

println

@Nonnull
public static <T> Observer<T> println()

Creates an observer with debugging purposes. It prints the submitted values to STDOUT with a line break, the exceptions to STDERR and prints an empty newline when it receives a finish().

Type Parameters:

T - the value type

Returns:

the observer

println

@Nonnull
public static <T> Observer<T> println(java.lang.String prefix)

Creates an observer with debugging purposes. It prints the submitted values to STDOUT with a line break, the exceptions to STDERR and prints an empty newline when it receives a finish().

Type Parameters:

T - the value type

Parameters:

prefix - the prefix to use when printing

Returns:

the observer

newAsyncAwaiter

@Nonnull
public static <T> Observer<T> newAsyncAwaiter(@Nonnull
                                                        java.lang.Runnable onDone)

Creates an observer which calls the supplied callback in case of an error() or finish() event.

Type Parameters:

T - the element type

Parameters:

onDone - the callback to invoke on completion

Returns:

the observer

newAsyncAwaiter0

@Nonnull
public static <T> Observer<T> newAsyncAwaiter0(@Nonnull
                                                         Action0 onDone)

Creates an observer which calls the supplied callback in case of an error() or finish() event.

Type Parameters:

T - the element type

Parameters:

onDone - the callback to invoke on completion

Returns:

the observer

newAsyncAwaiter

@Nonnull
public static <T> Observer<T> newAsyncAwaiter(@Nonnull
                                                        java.util.concurrent.CountDownLatch latch)

Creates an observer which counts down the latch by one in case of an error() or finish() event.

Type Parameters:

T - the element type

Parameters:

latch - the latch to count down by one

Returns:

the observer

newAsyncAwaiter

@Nonnull
public static <T> Observer<T> newAsyncAwaiter(@Nonnull
                                                        java.util.concurrent.locks.Lock lock,
                                                        @Nonnull
                                                        java.util.concurrent.locks.Condition cond)

Creates an observer which signals the condition in case of an error() or finish() event.

Type Parameters:

T - the element type

Parameters:

lock - the lock owning the condition

cond - the condition to signal

Returns:

the observer

newAsyncAwaiter

@Nonnull
public static <T> Observer<T> newAsyncAwaiter(@Nonnull
                                                        java.lang.Object syncObject)

Creates an observer which calls notifyAll on the supplied synchronization object in case of an error() or finish() event.

Type Parameters:

T - the element type

Parameters:

syncObject - the synchronization object

Returns:

the observer

select

@Nonnull
public static <T,U> Observer<U> select(@Nonnull
                                                 Observer<? super T> observer,
                                                 @Nonnull
                                                 Func1<? super U,? extends T> selector)

Modifies the source observer to use the selector to compute a new value from the received one.

The error and finish events are forwarded without modification.

Type Parameters:

T - the source element type

U - the result element type

Parameters:

observer - the observer to modify behavior on

selector - the selector to change the observed value

Returns:

the new observer

toObserver

@Nonnull
public static OriginalObserverWrapper toObserver(@Nonnull
                                                           java.util.Observer javaObserver,
                                                           @Nullable
                                                           java.util.Observable javaObservable)

Converts the original Java Observer into a reactive-observer.

The observer deregisters from the sender observable in case of error or finish.

Parameters:

javaObserver - the java observer

javaObservable - the sender java observable

Returns:

the reactive-observable

toJavaObserver

@Nonnull
public static <T> java.util.Observer toJavaObserver(Observer<T> observer)

Wraps the reactive-observer into a java-observer.

Note that since java-observer is non generic, ClassCastException might occur on the chain of reactive-observers in case an inproper object is propagated.

Type Parameters:

T - the element type

Parameters:

observer - the reactive-observer to wrap

Returns:

the java-observer

registerWith

@Nonnull
public static <T> java.io.Closeable registerWith(@Nonnull
                                                           java.util.Observable javaObservable,
                                                           @Nonnull
                                                           Observer<T> observer)

Convenience method to let a reactive-observer register with a java-observable.

The observer won't receive any error or finish events since java-observables don't have these concepts.

Note that java-observables are not generic, therefore ClassCastException might occur if inproper values are sent through the reactive-observable chain.

Type Parameters:

T - the element type

Parameters:

javaObservable - the observable used to register with.

observer - the observer to use

Returns:

the closeable to unregister the observer

registerWith

@Nonnull
public static java.io.Closeable registerWith(@Nonnull
                                                       java.util.Observable javaObservable,
                                                       @Nonnull
                                                       java.util.Observer javaObserver)

Convenience method to register a java-observer with a java-observable in the same style as the reactive counterpart do.

Note that java-observable prevents multiple registrations with the same java-observer instance, but there is no way to know if an observer is already registered. Therefore, in duplicate case the subsequent close handle will close the single registration.

Parameters:

javaObservable - the java-observable instance

javaObserver - the java-observer instance

Returns:

the close handle

registerWith

@Nonnull
public static <T> java.io.Closeable registerWith(@Nonnull
                                                           Observable<T> observable,
                                                           @Nonnull
                                                           java.util.Observer javaObserver)

Convenience method to register a java-observable with a reactive-observable.

Type Parameters:

T - the element type of the reactive-observable

Parameters:

observable - the reactive-observable to register with

javaObserver - the java-observer to register

Returns:

the close handler to deregister the observer

registerWith

@Nonnull
public static <T> java.io.Closeable registerWith(@Nonnull
                                                           java.lang.Iterable<T> iterable,
                                                           @Nonnull
                                                           Observer<? super T> observer)

"Registers" the observer with the iterable source and iterates over the source iterable in the current thread.

Type Parameters:

T - the element type

Parameters:

iterable - the iterable sequence

observer - the observer

Returns:

the close handler

registerWith

@Nonnull
public static <T> java.io.Closeable registerWith(@Nonnull
                                                           java.lang.Iterable<T> iterable,
                                                           @Nonnull
                                                           Observer<? super T> observer,
                                                           @Nonnull
                                                           Scheduler scheduler)

"Registers" an observer with the given iterable and iterates over it on the given scheduler. Each iteration is run as a new task.

Type Parameters:

T - the element type

Parameters:

iterable - the iterable source

observer - the observer

scheduler - the scheduler

Returns:

the closeable to cancel the operation

registerSafe

@Nonnull
public static <T> java.io.Closeable registerSafe(@Nonnull
                                                           Observable<? extends T> source,
                                                           @Nonnull
                                                           Observer<? super T> observer)

Registers the observer with the source in a way that if the registration call throws an exception, it is routed to the observer via the error() method.

Type Parameters:

T - the element type

Parameters:

source - the source observable

observer - the observer to register

Returns:

the close handle for the registration

Since:

0.97