com.diffplug.common.rx

Class Rx

java.lang.Object

com.diffplug.common.rx.Rx

public class Rx
extends Object

Unifies the listener models of RxJava's Observable with Guava's ListenableFuture , and also adds tracing capabilities.

TL;DR

 // subscribe to values, termination, or both
 Rx.subscribe(listenableOrObservable, val -> doSomething(val));                             // errors are passed to Errors.log()
 Rx.subscribe(listenableOrObservable, Rx.onTerminate(optionalError -> maybeHandleError());  // values are ignored
 Rx.subscribe(listenableOrObservable, Rx.onValueOrTerminate(val -> doSomething(val), optionalError -> maybeHandleError()));
 // receive callbacks on a specific executor
 Rx.on(someExecutor).subscribe(listenableOrObservable, val -> doSomething(val));
 // call unsubscribe() on the subscription to cancel it 
 rx.Subscription subscription = Rx.subscribe(listenableOrObservable, val -> doSomething);
 

Long version: `Rx` implements both the Observer and FutureCallback interfaces by mapping them to two `Consumer`s:

• `Consumer onValue`
• `Consumer> onTerminate`

Which are mapped as follows:

• `Observable.onNext(T value) -> onValue.accept(value)`
• `Observable.onCompleted() -> onTerminate.accept(Optional.empty())`
• `Observable.onError(Throwable error) -> onTerminate.accept(Optional.of(error))`
• `FutureCallback.onSuccess(T value) -> onValue.accept(value); onTerminate.accept(Optional.empty());`
• `FutureCallback.onError(Throwable error) -> onTerminate.accept(Optional.of(error))`

An instance of Rx is created by calling one of Rx's static creator methods:

• onValue(Consumer<T>)
• onTerminate(Consumer<Optional<Throwable>>)
• onFailure(Consumer<Throwable>)
• onValueOrTerminate(Consumer<T>, Consumer<Optional<Throwable>>)
• onValueOrFailure(Consumer<T>, Consumer<Throwable>)

Once you have an instance of Rx, you can subscribe it using the normal RxJava or Guava calls:

• `rxObservable.subscribe(Rx.onValue(val -> doSomething(val));`
• `Futures.addCallback(listenableFuture, Rx.onValue(val -> doSomething(val));`

But the recommended way to subscribe is to use:

• `Rx.subscribe(listenableOrObservable, Rx.onValue(val -> doSomething(val)));`
• `Rx.subscribe(listenableOrObservable, val -> doSomething(val)); // automatically uses Rx.onValue()`

The advantage of this latter method is that it returns Subscription instances which allow you to unsubscribe from futures in the same manner as for observables.

• `subscription = Rx.subscribe( ... )`

If you wish to receive callbacks on a specific thread, you can use:

• `Rx.on(someExecutor).subscribe( ... )`

Because RxJava's Observables use Schedulers rather than Executors, a Scheduler is automatically created using Schedulers. If you'd like to specify the Scheduler manually, you can use callbackOn(Executor, Scheduler) or you can create an executor which implements RxExecutor.Has.

See Also:

SwtExec

Method Summary

Modifier and Type	Method and Description
static RxExecutor	callbackOn(Executor executor) Mechanism for specifying a specific Executor.
static RxExecutor	callbackOn(Executor executor, Scheduler scheduler) Mechanism for specifying a specific Executor (for ListenableFuture) and Scheduler (for Observable).
static <T> Observable<T>	merge(IObservable<? extends T>... toMerge) Merges a bunch of IObservables into a single Observable containing the most-recent value.
static <T> RxListener<T>	onFailure(Consumer<Throwable> onFailure) Creates an Rx instance which will call the given consumer whenever the followed stream or future completes with an error.
static <T> RxListener<T>	onTerminate(Consumer<Optional<Throwable>> onTerminate) Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not.
static <T> RxListener<T>	onTerminateLogError(Consumer<Optional<Throwable>> onTerminate) Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not, and the error (if present) will be logged.
static <T> RxListener<T>	onValue(Consumer<T> onValue) Creates an Rx instance which will call the given consumer whenever a value is received.
static <T> RxListener<T>	onValueOnFailure(Consumer<T> onValue, Consumer<Throwable> onFailure) Creates an Rx instance which will call onValue whenever a value is received, and onFailure if the stream or future completes with an error.
static <T> RxListener<T>	onValueOnTerminate(Consumer<T> onValue, Consumer<Optional<Throwable>> onTerminate) Creates an Rx instance which will call onValue whenever a value is received, is received, and onTerminate when the future or observable completes, whether with an error or not.
static <T> RxListener<T>	onValueOnTerminateLogError(Consumer<T> onValue, Consumer<Optional<Throwable>> onTerminate) Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not, and the error (if present) will automatically be logged.
static RxExecutor	sameThreadExecutor()
static <T> Subscription	subscribe(CompletionStage<? extends T> future, Consumer<T> listener)
static <T> Subscription	subscribe(CompletionStage<? extends T> future, RxListener<T> listener)
static <T> Subscription	subscribe(IObservable<? extends T> observable, Consumer<T> listener)
static <T> Subscription	subscribe(IObservable<? extends T> observable, RxListener<T> listener)
static <T> Subscription	subscribe(ListenableFuture<? extends T> future, Consumer<T> listener)
static <T> Subscription	subscribe(ListenableFuture<? extends T> future, RxListener<T> listener)
static <T> Subscription	subscribe(Observable<? extends T> observable, Consumer<T> listener)
static <T> Subscription	subscribe(Observable<? extends T> observable, RxListener<T> listener)

Methods inherited from class java.lang.Object

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

Method Detail

onValue

public static <T> RxListener<T> onValue(Consumer<T> onValue)

Creates an Rx instance which will call the given consumer whenever a value is received. Any errors are sent to ErrorHandler.log().

onTerminate

public static <T> RxListener<T> onTerminate(Consumer<Optional<Throwable>> onTerminate)

Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not.

onTerminateLogError

public static <T> RxListener<T> onTerminateLogError(Consumer<Optional<Throwable>> onTerminate)

Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not, and the error (if present) will be logged.

onFailure

public static <T> RxListener<T> onFailure(Consumer<Throwable> onFailure)

Creates an Rx instance which will call the given consumer whenever the followed stream or future completes with an error.

onValueOnTerminate

public static <T> RxListener<T> onValueOnTerminate(Consumer<T> onValue,
                                                   Consumer<Optional<Throwable>> onTerminate)

Creates an Rx instance which will call onValue whenever a value is received, is received, and onTerminate when the future or observable completes, whether with an error or not.

onValueOnTerminateLogError

public static <T> RxListener<T> onValueOnTerminateLogError(Consumer<T> onValue,
                                                           Consumer<Optional<Throwable>> onTerminate)

Creates an Rx instance which will call the given consumer whenever the followed stream or future completes, whether with an error or not, and the error (if present) will automatically be logged.

onValueOnFailure

public static <T> RxListener<T> onValueOnFailure(Consumer<T> onValue,
                                                 Consumer<Throwable> onFailure)

Creates an Rx instance which will call onValue whenever a value is received, and onFailure if the stream or future completes with an error.

subscribe

public static <T> Subscription subscribe(Observable<? extends T> observable,
                                         RxListener<T> listener)

subscribe

public static <T> Subscription subscribe(Observable<? extends T> observable,
                                         Consumer<T> listener)

subscribe

public static <T> Subscription subscribe(IObservable<? extends T> observable,
                                         RxListener<T> listener)

subscribe

public static <T> Subscription subscribe(IObservable<? extends T> observable,
                                         Consumer<T> listener)

subscribe

public static <T> Subscription subscribe(ListenableFuture<? extends T> future,
                                         RxListener<T> listener)

subscribe

public static <T> Subscription subscribe(ListenableFuture<? extends T> future,
                                         Consumer<T> listener)

subscribe

public static <T> Subscription subscribe(CompletionStage<? extends T> future,
                                         RxListener<T> listener)

subscribe

public static <T> Subscription subscribe(CompletionStage<? extends T> future,
                                         Consumer<T> listener)

callbackOn

public static RxExecutor callbackOn(Executor executor)

Mechanism for specifying a specific Executor. A corresponding Scheduler will be created using Schedulers.from(executor).

callbackOn

public static RxExecutor callbackOn(Executor executor,
                                    Scheduler scheduler)

Mechanism for specifying a specific Executor (for ListenableFuture) and Scheduler (for Observable).

sameThreadExecutor

public static RxExecutor sameThreadExecutor()

merge

@SafeVarargs
public static <T> Observable<T> merge(IObservable<? extends T>... toMerge)

Merges a bunch of IObservables into a single Observable containing the most-recent value.