im trying to make publisher from specific scheduler.
i want the downstream operators to be on subscription thread by default (i.e if the callee didn’t specify the thread his operators would be run on the thread he calls subscribe from).

here is my code:

public class ReactorTest {
    private static final Scheduler SCH = Schedulers.newSingle("SCH-SINGLE");
    private int counter = 0;


    private void debugPrint(String functionName) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + functionName);
    }

    private int getCounter() {
        debugPrint("getCounter");
        return counter++;
    }

    private void trySleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {

        }
    }

    private int print(String str, int i) {
        trySleep(1000);
        debugPrint(str + " " + i);
        return i;
    }

    private Mono<Integer> getMono() {
        return Mono.fromCallable(this::getCounter)
                .publishOn(SCH)
                .publishOn(Schedulers.parallel());
    }

    public static void main(String[] args) throws InterruptedException {
        final ReactorTest m = new ReactorTest();
        for (int i = 0; i < 10; i++) {
            final int j = i;
            m.getMono()
                    .map(v -> m.print("map[" + j + "] ", v))
                    .subscribe(v -> m.print("subscribe[" + j + "] ", v));
        }
        Thread.sleep(5000);
        SCH.dispose();
    }
}

[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[SCH-SINGLE-1] getCounter
[parallel-6] map[5]  5
[parallel-9] map[8]  8
[parallel-8] map[7]  7
[parallel-2] map[1]  1
[parallel-10] map[9]  9
[parallel-1] map[0]  0
[parallel-3] map[2]  2
[parallel-7] map[6]  6
[parallel-5] map[4]  4
[parallel-4] map[3]  3
[parallel-9] subscribe[8]  8
[parallel-1] subscribe[0]  0
[parallel-10] subscribe[9]  9
[parallel-7] subscribe[6]  6
[parallel-3] subscribe[2]  2
[parallel-5] subscribe[4]  4
[parallel-8] subscribe[7]  7
[parallel-2] subscribe[1]  1
[parallel-4] subscribe[3]  3
[parallel-6] subscribe[5]  5

BUILD SUCCESSFUL in 5s
2 actionable tasks: 2 executed
12:15:04 PM: Execution finished ':ReactorTest.main()'.

this code would seperate my intended function to run on my specific thread. however i had to specify Schedulers.parallel or boundedelastic. i tried Schedulers.immediate() i thought it would pass the subscribing thread to be the one executing map and subscribe callback but it just passes the single scheduler i had which makes it congested because of the blocking user functions.

my solution was to use CompletableFuture and return Mono.fromFuture which would work but i just wanted to explore the operator way on reactor as its not very easy to work with composed CompletableFutures.

what would be the solution here ?

Also another question for following snippet:

Mono.fromCallable(this::getCounter) <------ (1)
                .publishOn(SCH)  <------ (2)
                .publishOn(Schedulers.parallel())
                .subscribeOn(Schedulers.parallel())  <------ (3)

why did (1) get called by (2) and not (3) here.