I validate my comment. block() checks if the calling thread is compatible with blocking code (thread external to reactor, or a thread of a specific reactor scheduler like Schedulers.boundedElastic()).
There's 2 ways to handle blocking calls in the middle of a reactive stack:
- Use the share operator on publisher you will block upon. Be careful, the share operator caches the value internally.
- Force
block() call to be executed on a blocking compatible scheduler using scheduleOn or publishOn. Beware, this operators should not be called on the publisher that directly call block(), but on the publisher that will "wrap" the block call (see example below).
Some references:
And a minimal reproducible example (tested on v3.4.6) giving this output:
Ok context: not running from reactor Threads
value is true
Problematic stack: working with scheduler not compatible with blocking call
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-2
Bad way to subscribe on a blocking compatible scheduler
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-4
Bad way to publish on blocking compatible scheduler
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-6
Possible workaround: share the reactive stream before blocking on it
It worked
Right way to subscribe on blocking compatible scheduler
It worked
Right way to publish on blocking compatible scheduler
It worked
Here comes the code:
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.util.concurrent.Callable;
import java.util.function.Supplier;
public class BlockingWorkaround {
public static void main(String[] args) throws Exception {
System.out.println("Ok context: not running from reactor Threads");
System.out.println("value is "+blockingFunction());
System.out.println("Problematic stack: working with scheduler not compatible with blocking call");
executeAndWait(() -> blockingFunction());
System.out.println("Bad way to subscribe on a blocking compatible scheduler");
executeAndWait(() -> blockingFunctionUsingSubscribeOn());
System.out.println("Bad way to publish on blocking compatible scheduler");
executeAndWait(() -> blockingFunctionUsingPublishOn());
System.out.println("Possible workaround: share the reactive stream before blocking on it");
executeAndWait(() -> blockingFunctionShared());
System.out.println("Right way to subscribe on blocking compatible scheduler");
subscribeOnAndWait(() -> blockingFunction());
System.out.println("Right way to publish on blocking compatible scheduler");
publishOnAndWait(() -> blockingFunction());
}
static Boolean blockingFunction() {
return delay()
.flatMap(delay -> Mono.just(true))
.block();
}
static Boolean blockingFunctionShared() {
return delay()
.flatMap(delay -> Mono.just(true))
.share() // Mono result is cached internally
.block();
}
static Boolean blockingFunctionUsingSubscribeOn() {
return delay()
.subscribeOn(Schedulers.boundedElastic())
.flatMap(delay -> Mono.just(true))
.block();
}
static Boolean blockingFunctionUsingPublishOn() {
return delay()
.flatMap(delay -> Mono.just(true))
.publishOn(Schedulers.boundedElastic())
.block();
}
static Mono<Long> delay() {
return Mono.delay(Duration.ofMillis(10));
}
private static void executeAndWait(Supplier<Boolean> blockingAction) throws InterruptedException {
delay()
.map(it -> blockingAction.get())
.subscribe(
val -> System.out.println("It worked"),
err -> System.out.println("ERROR: " + err.getMessage())
);
Thread.sleep(100);
}
private static void subscribeOnAndWait(Callable<Boolean> blockingAction) throws InterruptedException {
final Mono<Boolean> blockingMono = Mono.fromCallable(blockingAction)
.subscribeOn(Schedulers.boundedElastic()); // Upstream is executed on given scheduler
delay()
.flatMap(it -> blockingMono)
.subscribe(
val -> System.out.println("It worked"),
err -> System.out.println("ERROR: " + err.getMessage())
);
Thread.sleep(100);
}
private static void publishOnAndWait(Supplier<Boolean> blockingAction) throws InterruptedException {
delay()
.publishOn(Schedulers.boundedElastic()) // Cause downstream to be executed on given scheduler
.map(it -> blockingAction.get())
.subscribe(
val -> System.out.println("It worked"),
err -> System.out.println("ERROR: " + err.getMessage())
);
Thread.sleep(100);
}
}
blockin Schedulers not explicitly marked as compatible with blocking code. I do not find detailed doc about that, but I think that. you have to use eithersubscribeOn(Schedulers.boundedElastic())or.share()method on your flux before block call. The difference in behaviour might come from the fact that in one case, block function triggers pipeline in current thread (non reactive scheduler), but in your test, you try to callcanDoItfrom a reactive stack, so the executing thread comes from a reactive scheduler. – FantinlatoursubscribeOnandsharebut I still get the same error. Please note that the error does not appear withWebClient, it appears only withWebTestClient. – FrissellflatMapwould be nice. It would help me to dig further and adapt my answer. – Fantinlatour