SchedulerWhen extends Scheduler implements Disposable { @Override public void dispose() { disposable.dispose(); } SchedulerWhen(Function<Flowable<Flowable<Completable>>, Completable> combine, Scheduler actualScheduler); @Override void dispose(); @Override boolean isDisposed(); @NonNull @Override Worker createWorker();  }
@Test public void testAsyncMaxConcurrent() { TestScheduler tSched = new TestScheduler(); SchedulerWhen sched = maxConcurrentScheduler(tSched); TestSubscriber<Long> tSub = TestSubscriber.create(); asyncWork(sched).subscribe(tSub); tSub.assertValueCount(0); tSched.advanceTimeBy(0, SECONDS); tSub.assertValueCount(0); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(2); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(4); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(5); tSub.assertComplete(); sched.dispose(); }
@Test public void testAsyncDelaySubscription() { final TestScheduler tSched = new TestScheduler(); SchedulerWhen sched = throttleScheduler(tSched); TestSubscriber<Long> tSub = TestSubscriber.create(); asyncWork(sched).subscribe(tSub); tSub.assertValueCount(0); tSched.advanceTimeBy(0, SECONDS); tSub.assertValueCount(0); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(1); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(1); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(2); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(2); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(3); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(3); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(4); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(4); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(5); tSub.assertComplete(); sched.dispose(); }
@Test public void testSyncMaxConcurrent() { TestScheduler tSched = new TestScheduler(); SchedulerWhen sched = maxConcurrentScheduler(tSched); TestSubscriber<Long> tSub = TestSubscriber.create(); syncWork(sched).subscribe(tSub); tSub.assertValueCount(0); tSched.advanceTimeBy(0, SECONDS); tSub.assertValueCount(5); tSub.assertComplete(); sched.dispose(); }
@Test public void testSyncDelaySubscription() { final TestScheduler tSched = new TestScheduler(); SchedulerWhen sched = throttleScheduler(tSched); TestSubscriber<Long> tSub = TestSubscriber.create(); syncWork(sched).subscribe(tSub); tSub.assertValueCount(0); tSched.advanceTimeBy(0, SECONDS); tSub.assertValueCount(1); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(2); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(3); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(4); tSched.advanceTimeBy(1, SECONDS); tSub.assertValueCount(5); tSub.assertComplete(); sched.dispose(); }