observeOn unsubscribes child observables/interrupts the thread

[stannnous]

Hi,

The following test shows the use of observeOn in an observable chain interrupts the thread, cancelling child observables.

import org.junit.Test;
import rx.Completable;
import rx.schedulers.Schedulers;

import java.util.concurrent.CountDownLatch;

public class InterruptTest {
    @Test
    public void testInterrupt() throws InterruptedException {
       final int count = 1000;
       CountDownLatch latch = new CountDownLatch(count);

        for (int i = 0; i < count; i++) {
            Completable.complete()
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io()) // The problem does not occur if you comment out this line
                .andThen(Completable.fromAction(() -> {
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        System.out.println("Interrupted!"); // This is output periodically
                    }
                }))
                .subscribe(() -> latch.countDown());
        }

        latch.await();
        testInterrupt();
    }
}

The sleep is interrupted periodically printing "Interrupted!" while the test is running. The problem does not occur when the observeOn line is commented out.

We've tested this in RxJava 1.3.2 and 2.1.6 with the same results.

[akarnokd]

andThen and concat dispose the previous sources - which is unnecessary on its own - and thus a particular timing may cause the worker of the IO Scheduler to be disposed while it is serving the second Completable.

[akarnokd]

Fix in #5695.

[stannnous]

Hi @akarnokd thanks for the fast response. Is there any chance we can get this fixed in 1.x as well?

[akarnokd]

Yes, but it may take an additional week to have it released. Besides, we would like to encourage people to switch to v2 as it's only 6 months till RxJava 1.x' end-of-life.

[akarnokd]

Closing via #5696