BlockingOperatorToFuture { public static <T> Future<T> toFuture(Observable<? extends T> that) { final CountDownLatch finished = new CountDownLatch(1); final AtomicReference<T> value = new AtomicReference<T>(); final AtomicReference<Throwable> error = new AtomicReference<Throwable>(); final Subscription s = that.single().subscribe(new Subscriber<T>() { @Override public void onCompleted() { finished.countDown(); } @Override public void onError(Throwable e) { error.compareAndSet(null, e); finished.countDown(); } @Override public void onNext(T v) { value.set(v); } }); return new Future<T>() { private volatile boolean cancelled = false; @Override public boolean cancel(boolean mayInterruptIfRunning) { if (finished.getCount() > 0) { cancelled = true; s.unsubscribe(); finished.countDown(); return true; } else { return false; } } @Override public boolean isCancelled() { return cancelled; } @Override public boolean isDone() { return finished.getCount() == 0; } @Override public T get() throws InterruptedException, ExecutionException { finished.await(); return getValue(); } @Override public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (finished.await(timeout, unit)) { return getValue(); } else { throw new TimeoutException("Timed out after " + unit.toMillis(timeout) + "ms waiting for underlying Observable."); } } private T getValue() throws ExecutionException { if (error.get() != null) { throw new ExecutionException("Observable onError", error.get()); } else if (cancelled) { throw new CancellationException("Subscription unsubscribed"); } else { return value.get(); } } }; } private  BlockingOperatorToFuture(); static Future<T> toFuture(Observable<? extends T> that);  }
@Test public void testToFuture() throws InterruptedException, ExecutionException { Observable<String> obs = Observable.just("one"); Future<String> f = toFuture(obs); assertEquals("one", f.get()); }
@Test public void testToFutureList() throws InterruptedException, ExecutionException { Observable<String> obs = Observable.just("one", "two", "three"); Future<List<String>> f = toFuture(obs.toList()); assertEquals("one", f.get().get(0)); assertEquals("two", f.get().get(1)); assertEquals("three", f.get().get(2)); }
@Test(expected = IllegalArgumentException.class) public void testExceptionWithMoreThanOneElement() throws Throwable { Observable<String> obs = Observable.just("one", "two"); Future<String> f = toFuture(obs); try { f.get(); } catch(ExecutionException e) { throw e.getCause(); } }
@Test public void testToFutureWithException() { Observable<String> obs = Observable.create(new OnSubscribe<String>() { @Override public void call(Subscriber<? super String> observer) { observer.onNext("one"); observer.onError(new TestException()); } }); Future<String> f = toFuture(obs); try { f.get(); fail("expected exception"); } catch (Throwable e) { assertEquals(TestException.class, e.getCause().getClass()); } }
@Test(expected=CancellationException.class) public void testGetAfterCancel() throws Exception { Observable<String> obs = Observable.create(new OperationNeverComplete<String>()); Future<String> f = toFuture(obs); boolean cancelled = f.cancel(true); assertTrue(cancelled); f.get(); }
@Test(expected=CancellationException.class) public void testGetWithTimeoutAfterCancel() throws Exception { Observable<String> obs = Observable.create(new OperationNeverComplete<String>()); Future<String> f = toFuture(obs); boolean cancelled = f.cancel(true); assertTrue(cancelled); f.get(Long.MAX_VALUE, TimeUnit.NANOSECONDS); }
@Test(expected = NoSuchElementException.class) public void testGetWithEmptyObservable() throws Throwable { Observable<String> obs = Observable.empty(); Future<String> f = obs.toBlocking().toFuture(); try { f.get(); } catch(ExecutionException e) { throw e.getCause(); } }
@Test public void testGetWithASingleNullItem() throws Exception { Observable<String> obs = Observable.just((String)null); Future<String> f = obs.toBlocking().toFuture(); assertEquals(null, f.get()); }