SubscribeOn or PublishOn not working with ReactiveCassandraCrudRepository

Question

We're using Reactive Spring Data Repository with Spring WebFlux, My understanding for SubscribeOn says it decides on which ThreadPool the operators before SubscribeOn will execute in the flux, while PublishOn decides the actual ThreadPool on which the subscription will execute. However in the below code even with PublishOn and SubscribeOn, the code isn't getting executed on Main Thread rather it's falling back to Cluster-nio-worker-1.

    System.out.println("Current Thread :- "+Thread.currentThread().getName()); //Current Thread :- main
    personRepository.findAll().log()
            .map(document -> mapDocumentToSomethingElse(document)) //Current thread cluster-nio-worker-1
            .subscribeOn(Schedulers.immediate())
            .publishOn(Schedulers.immediate())
            .subscribe(trackingevent -> System.out.println("Got Item "+item +" inside thread "+Thread.currentThread()), //Thread[cluster-nio-worker-1,5,main]
                    excp -> excp.printStackTrace(),
                    () -> System.out.println("Completed processing Thread:- "+Thread.currentThread().getName())); //cluster-nio-worker-1

Also what does Thread[cluster-nio-worker-1,5,main] mean? Why are these method calls not using the main thread for execution.

vins vins · Accepted Answer · 2020-04-23T18:20:56

The subscribeOn method makes the publisher to use the given thread pool for publishing the values. There could be N number of subscribeOn method in the pipeline. The cloest one will take effect. personRepository.findAll().log() is a wrapper and returns a flux. So if it uses any Schedulers internally, then you can not change it using the subscribeOn. For example, interval method uses parallel and I can not change it to boundedElastic as shown here.

    Flux.interval(Duration.ofSeconds(1))
            .subscribeOn(Schedulers.boundedElastic())

Schedulers.immediate just keeps the pipeline execution in the same thread. It is not main and in your case it will be cluster-nio-worker thread pool.

We can switch from main to any Schedulers thread pool. But we can not switch the execution back to main thread. It is not a limitation of project reactor. It should be a limitation of Java itself.

SubscribeOn or PublishOn not working with ReactiveCassandraCrudRepository

1 Answers