SingleScheduler extends Scheduler { @NonNull @Override public Disposable scheduleDirect(@NonNull Runnable run, long delay, TimeUnit unit) { ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run)); try { Future<?> f; if (delay <= 0L) { f = executor.get().submit(task); } else { f = executor.get().schedule(task, delay, unit); } task.setFuture(f); return task; } catch (RejectedExecutionException ex) { RxJavaPlugins.onError(ex); return EmptyDisposable.INSTANCE; } } SingleScheduler();  SingleScheduler(ThreadFactory threadFactory); @Override void start(); @Override void shutdown(); @NonNull @Override Worker createWorker(); @NonNull @Override Disposable scheduleDirect(@NonNull Runnable run, long delay, TimeUnit unit); @NonNull @Override Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initialDelay, long period, TimeUnit unit);  }
@Test(timeout = 1000) public void runnableDisposedAsync() throws Exception { final Scheduler s = Schedulers.single(); Disposable d = s.scheduleDirect(Functions.EMPTY_RUNNABLE); while (!d.isDisposed()) { Thread.sleep(1); } }
@Test(timeout = 1000) public void runnableDisposedAsyncCrash() throws Exception { final Scheduler s = Schedulers.single(); Disposable d = s.scheduleDirect(new Runnable() { @Override public void run() { throw new IllegalStateException(); } }); while (!d.isDisposed()) { Thread.sleep(1); } }
@Test(timeout = 1000) public void runnableDisposedAsyncTimed() throws Exception { final Scheduler s = Schedulers.single(); Disposable d = s.scheduleDirect(Functions.EMPTY_RUNNABLE, 1, TimeUnit.MILLISECONDS); while (!d.isDisposed()) { Thread.sleep(1); } }