BlockingSubscriber extends AtomicReference<Subscription> implements FlowableSubscriber<T>, Subscription { @Override public void cancel() { if (SubscriptionHelper.cancel(this)) { queue.offer(TERMINATED); } } BlockingSubscriber(Queue<Object> queue); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void request(long n); @Override void cancel(); boolean isCancelled(); static final Object TERMINATED; }
@Test public void cancel() { BlockingSubscriber<Integer> bq = new BlockingSubscriber<Integer>(new ArrayDeque<Object>()); assertFalse(bq.isCancelled()); bq.cancel(); assertTrue(bq.isCancelled()); bq.cancel(); assertTrue(bq.isCancelled()); }