Observable from Java Future

[benjchristensen]

Great post at http://www.datastax.com/dev/blog/java-driver-async-queries. I wanted to share an improvement that can be made. Using Observable.from(Future f) is unfortunately limited in what it can do because the Java future doesn't have a callback. All that can be done is either block on future.get() or poll future.get(), both of which are inefficient and not push based.

It would be preferable if the Datastax driver exposed an actual callback that could be hooked into. Without that it is preferable to hook into the ListenableFuture as it allows registering a listener.

You can see code at https://github.com/ReactiveX/RxJavaGuava/blob/master/src/main/java/rx/observable/ListenableFutureObservable.java#L51

    public static <T> Observable<T> from(final ListenableFuture<T> future, final Executor executor) {
        return Observable.create(new OnSubscribe<T>() {
            @Override
            public void call(final Subscriber<? super T> subscriber) {
                future.addListener(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            T t = future.get();
                            subscriber.onNext(t);
                            subscriber.onCompleted();
                        } catch (Exception e) {
                            subscriber.onError(e);
                        }
                    }
                }, executor);
            }
        });
    }

This code is not yet released and may have further improvements to it, but this achieves the desired goal of being non-blocking and only reacting once the Future is done and calls back with the data.