TrampolineScheduler extends Scheduler { @Override public Worker createWorker() { return new InnerCurrentThreadScheduler(); } TrampolineScheduler(); @Override Worker createWorker();  }
@Test public void testNestedTrampolineWithUnsubscribe() { final ArrayList<String> workDone = new ArrayList<String>(); final CompositeSubscription workers = new CompositeSubscription(); Worker worker = Schedulers.trampoline().createWorker(); try { workers.add(worker); worker.schedule(new Action0() { @Override public void call() { workers.add(doWorkOnNewTrampoline("A", workDone)); } }); final Worker worker2 = Schedulers.trampoline().createWorker(); workers.add(worker2); worker2.schedule(new Action0() { @Override public void call() { workers.add(doWorkOnNewTrampoline("B", workDone)); worker2.unsubscribe(); } }); assertEquals(6, workDone.size()); assertEquals(Arrays.asList("A.1", "A.B.1", "A.B.2", "B.1", "B.B.1", "B.B.2"), workDone); } finally { workers.unsubscribe(); } }
@Test public void testTrampolineWorkerHandlesConcurrentScheduling() { final Worker trampolineWorker = Schedulers.trampoline().createWorker(); final Observer<Subscription> observer = Observers.empty(); final TestSubscriber<Subscription> ts = new TestSubscriber<Subscription>(observer); Observable.range(0, 50) .flatMap(new Func1<Integer, Observable<Subscription>>() { @Override public Observable<Subscription> call(Integer count) { return Observable.interval(1, TimeUnit.MICROSECONDS).map( new Func1<Long, Subscription>() { @Override public Subscription call(Long count) { return trampolineWorker.schedule(new Action0() { @Override public void call() {} }); } }).limit(100); } }) .subscribeOn(Schedulers.computation()) .subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); }