@Test public void onSuccessCancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishSubject<Integer> ps = PublishSubject.create(); final Future<?> f = ps.single(-99).toFuture(); ps.onNext(1); Runnable r1 = new Runnable() { @Override public void run() { f.cancel(true); } }; Runnable r2 = new Runnable() { @Override public void run() { ps.onComplete(); } }; TestHelper.race(r1, r2); } }
@Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { return false; } if (s.compareAndSet(a, DisposableHelper.DISPOSED)) { if (a != null) { a.dispose(); } countDown(); return true; } } }
FutureSingleObserver extends CountDownLatch implements SingleObserver<T>, Future<T>, Disposable { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { return false; } if (s.compareAndSet(a, DisposableHelper.DISPOSED)) { if (a != null) { a.dispose(); } countDown(); return true; } } } }
FutureSingleObserver extends CountDownLatch implements SingleObserver<T>, Future<T>, Disposable { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { return false; } if (s.compareAndSet(a, DisposableHelper.DISPOSED)) { if (a != null) { a.dispose(); } countDown(); return true; } } } FutureSingleObserver(); }
FutureSingleObserver extends CountDownLatch implements SingleObserver<T>, Future<T>, Disposable { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { return false; } if (s.compareAndSet(a, DisposableHelper.DISPOSED)) { if (a != null) { a.dispose(); } countDown(); return true; } } } FutureSingleObserver(); @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 onSuccess(T t); @Override void onError(Throwable t); @Override void dispose(); @Override boolean isDisposed(); }
FutureSingleObserver extends CountDownLatch implements SingleObserver<T>, Future<T>, Disposable { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Disposable a = s.get(); if (a == this || a == DisposableHelper.DISPOSED) { return false; } if (s.compareAndSet(a, DisposableHelper.DISPOSED)) { if (a != null) { a.dispose(); } countDown(); return true; } } } FutureSingleObserver(); @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 onSuccess(T t); @Override void onError(Throwable t); @Override void dispose(); @Override boolean isDisposed();  }