@Test public void cancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final FutureSubscriber<Integer> fs = new FutureSubscriber<Integer>(); Runnable r = new Runnable() { @Override public void run() { fs.cancel(false); } }; TestHelper.race(r, r); } }
@Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Subscription a = s.get(); if (a == this || a == SubscriptionHelper.CANCELLED) { return false; } if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) { if (a != null) { a.cancel(); } countDown(); return true; } } }
FutureSubscriber extends CountDownLatch implements FlowableSubscriber<T>, Future<T>, Subscription { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Subscription a = s.get(); if (a == this || a == SubscriptionHelper.CANCELLED) { return false; } if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) { if (a != null) { a.cancel(); } countDown(); return true; } } } }
FutureSubscriber extends CountDownLatch implements FlowableSubscriber<T>, Future<T>, Subscription { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Subscription a = s.get(); if (a == this || a == SubscriptionHelper.CANCELLED) { return false; } if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) { if (a != null) { a.cancel(); } countDown(); return true; } } } FutureSubscriber(); }
FutureSubscriber extends CountDownLatch implements FlowableSubscriber<T>, Future<T>, Subscription { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Subscription a = s.get(); if (a == this || a == SubscriptionHelper.CANCELLED) { return false; } if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) { if (a != null) { a.cancel(); } countDown(); return true; } } } FutureSubscriber(); @Override boolean cancel(boolean mayInterruptIfRunning); @Override boolean isCancelled(); @Override boolean isDone(); @Override T get(); @Override T get(long timeout, TimeUnit unit); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void cancel(); @Override void request(long n); }
FutureSubscriber extends CountDownLatch implements FlowableSubscriber<T>, Future<T>, Subscription { @Override public boolean cancel(boolean mayInterruptIfRunning) { for (;;) { Subscription a = s.get(); if (a == this || a == SubscriptionHelper.CANCELLED) { return false; } if (s.compareAndSet(a, SubscriptionHelper.CANCELLED)) { if (a != null) { a.cancel(); } countDown(); return true; } } } FutureSubscriber(); @Override boolean cancel(boolean mayInterruptIfRunning); @Override boolean isCancelled(); @Override boolean isDone(); @Override T get(); @Override T get(long timeout, TimeUnit unit); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void cancel(); @Override void request(long n);  }