BlockingObserver extends AtomicReference<Disposable> implements Observer<T>, Disposable { @Override public void dispose() { if (DisposableHelper.dispose(this)) { queue.offer(TERMINATED); } } BlockingObserver(Queue<Object> queue); @Override void onSubscribe(Disposable s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void dispose(); @Override boolean isDisposed(); static final Object TERMINATED; }
@Test public void dispose() { Queue<Object> q = new ArrayDeque<Object>(); BlockingObserver<Object> bo = new BlockingObserver<Object>(q); bo.dispose(); assertEquals(BlockingObserver.TERMINATED, q.poll()); bo.dispose(); assertNull(q.poll()); }