Reactor-core: Mono.fromFuture does not respect subscribeOn schedulers

[cquludajun]

Expected behavior

Mono object should always respect schedulers defined in subscribeOn method

Actual behavior

running in the completable future's thread pool

Steps to reproduce

    @Test
    public void monoFromCompletableFutureDoesNotRespectSubscribeOnSchedulers() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            sleep(1000);   // sleep to make sure it won't complete immediately, callbacks in completable future will run in default ForkJoin thread pool
            System.out.println("CompletableFuture supply async running in : " + Thread.currentThread().getName());
            return Thread.currentThread().getName();
        });


        Mono.fromFuture(future)
                .doOnNext(s -> System.out.println("doOnNext is not running in new-parallel thread pool but in " + Thread.currentThread().getName()))
                .publishOn(Schedulers.newElastic("new-elastic"))  // if you comment this out, the following map and subscribe will run in ForkJoinPool
                .map(s -> {
                    System.out.println("map running in thread " + getThreadName());
                    return s;
                })
                .subscribeOn(Schedulers.newParallel("new-parallel"))
                .subscribe(s -> {
                    System.out.println("subscribe running in thread " + Thread.currentThread().getName());
                    latch.countDown();
                });

        System.out.println("I am main and waiting...");

        latch.await();
    }

    private void sleep(long miliseconds) {
        try {
            Thread.sleep(miliseconds);
        } catch (Exception e) {
        }
    }

Reactor Core version

3.2.2.RELEASE

JVM version (e.g. java -version)

$ java -version
java version "1.8.0_91"
Java(TM) SE Runtime Environment (build 1.8.0_91-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.91-b14, mixed mode)

[cquludajun]

It is not a problem in RXJava2:

    @Test
    public void testRxJava2FromFuture() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            System.out.println("supplyAsync is running in " + Thread.currentThread().getName());
            return Thread.currentThread().getName();
        });

        Observable<String> observable = Observable.fromFuture(future);

        observable
                .doOnNext(s -> System.out.println("doOnNext is running in " + Thread.currentThread().getName()))
                .map(s -> {
                    System.out.println("map is running in " + Thread.currentThread().getName());
                    return s;
                })
                .subscribeOn(io.reactivex.schedulers.Schedulers.io())
                .subscribe(s -> {
                    System.out.println("subscribe is running in " +  Thread.currentThread().getName());
                    latch.countDown();
                });

        System.out.println("I am main and waiting...");

        latch.await();
    }

[akarnokd]

Not sure why you expected subscribeOn to take any effect. supplyAsync will run the body on a Java-specific threadpool and complete there, triggering Reactor/RxJava to process its result on that same caller thread, unless you specify publishOn/observeOn to move it to one of the Reactor/RxJava specific thread.

[akarnokd]

Also there is a difference between Reactor's implementation and RxJava's implementation of fromFuture. RxJava blocks as part of the subscription process hence subscribeOn has meaning for it.

Reactor's fromFuture doesn't block but continues non-blockingly and it works with CompletableFutures that actually allow this non-blocking behavior. RxJava is Java 6 only thus it can't support CompletableFuture by itself.

[cquludajun]

I thought subscribeOn would always change the subscription thread. That was the reason I came out of this test case. It looks like subscribeOn has its limitations. Is this correct? Do you have some other examples/docs so that I could better understand subscribeOn?

[simonbasle]

You can see it this way: the adapter needs to delegate to the CompletableFuture's execution model for the general case (no subscribeOn involved). So what it does is simply attach a listener at subscription.
As a result, subscribeOn only impacts the thread _from_ which the listener is attached. After that, the CompletableFuture uses its own thread when relevant methods are called.