2.x: parallel blocks forever with TestScheduler

[asarkar]

io.reactivex.rxjava2:rxjava:2.2.3

public class DemoApplication {
   public static void main(String[] args) {
       List<Map.Entry<Integer, String>> entries = Arrays.asList(
               new AbstractMap.SimpleImmutableEntry<>(3, "hello"),
               new AbstractMap.SimpleImmutableEntry<>(3, "world")
       );

       Flowable.fromIterable(entries)
               .flatMap(entry -> Flowable.range(1, entry.getKey())
                       .parallel()
                       .runOn(Schedulers.io())
                       .flatMap(i -> logIt(entry.getValue()).toFlowable())
                       .sequential()
               )
               .all(success -> success)
               .doOnSuccess(success -> System.out.println("good"))
               .doOnError(Throwable::printStackTrace)
               .blockingGet();
   }

   private static Single<Boolean> logIt(String s) {
       System.out.println(s);

       return Single.just(true);
   }
}

This code works as expected. However, replacing Schedulers.io() with TestScheduler blocks forever. Gradle project attached.

demo.zip

[akarnokd]

Do you call advanceTimeBy anywhere?

[asarkar]

Nop, am I supposed to? I'm under the impression that the work only depends on availability of threads, and no delay is involved. Also, AFAIK, TestScheduler is single threaded, so all I'm expecting is the messages logged sequentially.

[akarnokd]

There are plenty of resources on the internet, such as this that demonstrate TestScheduler. You can find other resources by searching for RxJava TestScheduler in your favorite search engine.

[asarkar]

There are exactly zero resources on the internet that demonstrate TestScheduler along with parallel. I couldn't find other resources for the same in my favorite search engine.

[akarnokd]

Schedulers are orthogonal to flows, so conceptually it doesn't matter if you run a sequential or parallel flow with them.

[asarkar]

Hmm, so TestScheduler doesn't start until the time is advanced, even if there's no delay or interval in the code (nothing that apparently involves "time"). That wouldn't be my first thought; perhaps you can add a note to the docs you linked to above to clarify this behavior, and why's it this way? Or, if you do that here, I'll be happy to send a PR for the docs.

[akarnokd]

even if there's no delay or interval in the code [...] That wouldn't be my first thought

TestScheduler captures tasks scheduled by operators so that side-effects can be performed before the task then gets executed on demand via advanceTimeBy, even if they are logically not delayed. In reality, there is always some non-zero time elapsing between the scheduling and execution of such task. TestScheduler just makes it deterministic.

if you do that here, I'll be happy to send a PR for the docs.

If you wish to contribute, you could extend its JavaDocs with the explicit mention of advanceTimeBy or even a short example.

[asarkar]

@akarnokd #6356