@Test public void onError() throws Exception { List<Throwable> errors = TestHelper.trackPluginErrors(); try { fo.onError(new TestException("One")); fo.onError(new TestException("Two")); try { fo.get(5, TimeUnit.MILLISECONDS); } catch (ExecutionException ex) { assertTrue(ex.toString(), ex.getCause() instanceof TestException); assertEquals("One", ex.getCause().getMessage()); } TestHelper.assertUndeliverable(errors, 0, TestException.class, "Two"); } finally { RxJavaPlugins.reset(); } }
@Override public void onError(Throwable t) { if (error == null) { error = t; for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { RxJavaPlugins.onError(t); return; } if (s.compareAndSet(a, this)) { countDown(); return; } } } else { RxJavaPlugins.onError(t); } }
FutureObserver extends CountDownLatch implements Observer<T>, Future<T>, Disposable { @Override public void onError(Throwable t) { if (error == null) { error = t; for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { RxJavaPlugins.onError(t); return; } if (s.compareAndSet(a, this)) { countDown(); return; } } } else { RxJavaPlugins.onError(t); } } }
FutureObserver extends CountDownLatch implements Observer<T>, Future<T>, Disposable { @Override public void onError(Throwable t) { if (error == null) { error = t; for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { RxJavaPlugins.onError(t); return; } if (s.compareAndSet(a, this)) { countDown(); return; } } } else { RxJavaPlugins.onError(t); } } FutureObserver(); }
FutureObserver extends CountDownLatch implements Observer<T>, Future<T>, Disposable { @Override public void onError(Throwable t) { if (error == null) { error = t; for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { RxJavaPlugins.onError(t); return; } if (s.compareAndSet(a, this)) { countDown(); return; } } } else { RxJavaPlugins.onError(t); } } FutureObserver(); @Override boolean cancel(boolean mayInterruptIfRunning); @Override boolean isCancelled(); @Override boolean isDone(); @Override T get(); @Override T get(long timeout, TimeUnit unit); @Override void onSubscribe(Disposable s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void dispose(); @Override boolean isDisposed(); }
FutureObserver extends CountDownLatch implements Observer<T>, Future<T>, Disposable { @Override public void onError(Throwable t) { if (error == null) { error = t; for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { RxJavaPlugins.onError(t); return; } if (s.compareAndSet(a, this)) { countDown(); return; } } } else { RxJavaPlugins.onError(t); } } FutureObserver(); @Override boolean cancel(boolean mayInterruptIfRunning); @Override boolean isCancelled(); @Override boolean isDone(); @Override T get(); @Override T get(long timeout, TimeUnit unit); @Override void onSubscribe(Disposable s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void dispose(); @Override boolean isDisposed();  }