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 
 io.reactivex.disposables.Disposable subscription = Rx.subscribe(listenableOrObservable, val -> doSomething);
 

Long version: `Rx` implements both the rx.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 Disposable 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 rx.Schedulers rather than Executors, a Scheduler is automatically created using rx.Schedulers#from. 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 RxTracingPolicy	getTracingPolicy() Returns the global tracing policy.
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> void	subscribe(CompletionStage<? extends T> future, Consumer<T> listener)
static <T> void	subscribe(CompletionStage<? extends T> future, RxListener<T> listener)
static <T> void	subscribe(IObservable<? extends T> observable, Consumer<T> listener)
static <T> void	subscribe(IObservable<? extends T> observable, RxListener<T> listener)
static <T> void	subscribe(ListenableFuture<? extends T> future, Consumer<T> listener)
static <T> void	subscribe(ListenableFuture<? extends T> future, RxListener<T> listener)
static <T> void	subscribe(Observable<? extends T> observable, Consumer<T> listener)
static <T> void	subscribe(Observable<? extends T> observable, RxListener<T> listener)
static <T> Disposable	subscribeDisposable(CompletionStage<? extends T> future, Consumer<T> listener)
static <T> Disposable	subscribeDisposable(CompletionStage<? extends T> future, RxListener<T> listener)
static <T> Disposable	subscribeDisposable(IObservable<? extends T> observable, Consumer<T> listener)
static <T> Disposable	subscribeDisposable(IObservable<? extends T> observable, RxListener<T> listener)
static <T> Disposable	subscribeDisposable(ListenableFuture<? extends T> future, Consumer<T> listener)
static <T> Disposable	subscribeDisposable(ListenableFuture<? extends T> future, RxListener<T> listener)
static <T> Disposable	subscribeDisposable(Observable<? extends T> observable, Consumer<T> listener)
static <T> Disposable	subscribeDisposable(Observable<? extends T> observable, RxListener<T> listener)
static <T> void	sync(RxBox<T> left, RxBox<T> right) Reliable way to sync two RxBox to each other.
static <T> void	sync(RxSubscriber subscriber, RxBox<T> left, RxBox<T> right) Reliable way to sync two RxBox to each other, using the given RxSubscriber to listen for changes

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> void subscribe(Observable<? extends T> observable,
                                 RxListener<T> listener)

subscribe

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

subscribe

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

subscribe

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

subscribe

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

subscribe

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

subscribe

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

subscribe

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

subscribeDisposable

public static <T> Disposable subscribeDisposable(Observable<? extends T> observable,
                                                 RxListener<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(Observable<? extends T> observable,
                                                 Consumer<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(IObservable<? extends T> observable,
                                                 RxListener<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(IObservable<? extends T> observable,
                                                 Consumer<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(ListenableFuture<? extends T> future,
                                                 RxListener<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(ListenableFuture<? extends T> future,
                                                 Consumer<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(CompletionStage<? extends T> future,
                                                 RxListener<T> listener)

subscribeDisposable

public static <T> Disposable subscribeDisposable(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()

getTracingPolicy

public static RxTracingPolicy getTracingPolicy()

Returns the global tracing policy.

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.

sync

public static <T> void sync(RxBox<T> left,
                            RxBox<T> right)

Reliable way to sync two RxBox to each other.

sync

public static <T> void sync(RxSubscriber subscriber,
                            RxBox<T> left,
                            RxBox<T> right)

Reliable way to sync two RxBox to each other, using the given RxSubscriber to listen for changes