BlockingSubscriber extends AtomicReference<Subscription> implements FlowableSubscriber<T>, Subscription { @Override public void onSubscribe(Subscription s) { if (SubscriptionHelper.setOnce(this, s)) { queue.offer(NotificationLite.subscription(this)); } } 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(timeout = 5000) public void blockingFirstTimeout2() { BlockingFirstSubscriber<Integer> bf = new BlockingFirstSubscriber<Integer>(); bf.onSubscribe(new BooleanSubscription()); Thread.currentThread().interrupt(); try { bf.blockingGet(); fail("Should have thrown!"); } catch (RuntimeException ex) { assertTrue(ex.toString(), ex.getCause() instanceof InterruptedException); } }
@Test public void cancelOnRequest() { final BlockingFirstSubscriber<Integer> bf = new BlockingFirstSubscriber<Integer>(); final AtomicBoolean b = new AtomicBoolean(); Subscription s = new Subscription() { @Override public void request(long n) { bf.cancelled = true; } @Override public void cancel() { b.set(true); } }; bf.onSubscribe(s); assertTrue(b.get()); }
@Test public void cancelUpfront() { final BlockingFirstSubscriber<Integer> bf = new BlockingFirstSubscriber<Integer>(); final AtomicBoolean b = new AtomicBoolean(); bf.cancelled = true; Subscription s = new Subscription() { @Override public void request(long n) { b.set(true); } @Override public void cancel() { } }; bf.onSubscribe(s); assertFalse(b.get()); }