QueueDrainObserver extends QueueDrainSubscriberPad2 implements Observer<T>, ObservableQueueDrain<U, V> { @Override public final boolean enter() { return wip.getAndIncrement() == 0; } QueueDrainObserver(Observer<? super V> actual, SimplePlainQueue<U> queue); @Override final boolean cancelled(); @Override final boolean done(); @Override final boolean enter(); final boolean fastEnter(); @Override final Throwable error(); @Override final int leave(int m); @Override void accept(Observer<? super V> a, U v);  }
@Test public void unorderedSlowPath() { TestObserver<Integer> to = new TestObserver<Integer>(); Disposable d = Disposables.empty(); QueueDrainObserver<Integer, Integer, Integer> qd = createUnordered(to, d); to.onSubscribe(Disposables.empty()); qd.enter(); qd.onNext(1); to.assertEmpty(); }
@Test public void orderedSlowPath() { TestObserver<Integer> to = new TestObserver<Integer>(); Disposable d = Disposables.empty(); QueueDrainObserver<Integer, Integer, Integer> qd = createOrdered(to, d); to.onSubscribe(Disposables.empty()); qd.enter(); qd.onNext(1); to.assertEmpty(); }