@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()); }
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(); } } }; }
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(); } } }; } }
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(); } } }; }  }
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); }
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);  }