### Question:
OperationToFuture { 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.subscribe(new Observer<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) { if (!value.compareAndSet(null, v)) { error.compareAndSet(null, new IllegalStateException("Observable.toFuture() only supports sequences with a single value. Use .toList().toFuture() if multiple values are expected.")); finished.countDown(); } } }); 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 { return value.get(); } } }; }  static Future<T> toFuture(Observable<? extends T> that);  }### Answer:
@Test public void testToFuture() throws InterruptedException, ExecutionException { Observable<String> obs = Observable.from("one"); Future<String> f = toFuture(obs); assertEquals("one", f.get()); }
@Test public void testToFutureList() throws InterruptedException, ExecutionException { Observable<String> obs = Observable.from("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 = ExecutionException.class) public void testExceptionWithMoreThanOneElement() throws InterruptedException, ExecutionException { Observable<String> obs = Observable.from("one", "two"); Future<String> f = toFuture(obs); assertEquals("one", f.get()); }
@Test public void testToFutureWithException() { Observable<String> obs = Observable.create(new OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { observer.onNext("one"); observer.onError(new TestException()); return Subscriptions.empty(); } }); Future<String> f = toFuture(obs); try { f.get(); fail("expected exception"); } catch (Throwable e) { assertEquals(TestException.class, e.getCause().getClass()); } }