2.0.4: an infinite disposable

[konmik]

This fails:

@Test
public void directScheduleOnSingleThreadExecutor() {
    Scheduler scheduler = Schedulers.from(Executors.newSingleThreadExecutor());

    Disposable disposable = scheduler.scheduleDirect(() -> {
    });

    long start = nanoTime();
    while (!disposable.isDisposed()) {
        assertTrue(nanoTime() - start < SECONDS.toNanos(10));
    }
}

while replacing from(...) with io() works

I can't find where the error exactly is, my debugger can't jump in to the right source code line.

[akarnokd]

Yes, the internal ExecutorScheduler.BooleanDisposable doesn't set its state to disposed after the task has run.

How much problem is it for you?

(Sidenote: It is generally not recommended to spin on isDisposed() because that nullifies the reason RxJava exists: not blocking on Future.get() or spinning on Future.isDone().)

[konmik]

The bug it was causing is that when user scrolls messages up, I wanted paging to run if it is not already running. After the transition to RxJava 2 I started to use this new scheduleDirect method. The first paging iteration worked as expected, but after that it just hangs up without any signs of living.

This is my current workaround:

public static Disposable scheduleDirect(Scheduler scheduler, Runnable runnable) {
    Scheduler.Worker worker = scheduler.createWorker();
    worker.schedule(() -> {
        try {
            runnable.run();
        } finally {
            worker.dispose();
        }
    });
    return worker;
}

[konmik]

(As a sidenote: some real-world tasks DO require blocking operations or running of tasks depending on other tasks state.)

[akarnokd]

Okay, for consistency with other Schedulers, I'll post a fix for this.

[konmik]

Cool! :)

[akarnokd]

Closing via #5005.