Is Mono.toFuture() blocking?

Viewed 4301

From the Official Documentation of Mono#block() it is said that:

Subscribe to this Mono and block indefinitely until a next signal is received. Returns that value, or null if the Mono completes empty. In case the Mono errors, the original exception is thrown (wrapped in a RuntimeException if it was a checked exception).

So it is sure that block() method is blocking and it will not execute the next line untill block() resolved.

But my confusion is while I was using toFuture() expecting it will be non-blocking but it is behaving exactly like block method. And in the Documentation of Mono#toFuture() it is stated:

Transform this Mono into a CompletableFuture completing on onNext or onComplete and failing on onError.

Mono#toFuture()

Not much clear. Nowhere in this doc said Mono#toFuture() is blocking.

  1. Please confirm me if toFuture() method blocking or non-blocking?
  2. Also If it is non-blocking then, which thread will responsible to execute the code inside CompletableFuture?

Update: added code snippet

using Mono.block() method:

    long time = System.currentTimeMillis();
    String block = Mono.fromCallable(() -> {
        logger.debug("inside in fromCallable() block()");
        //Upstream httpcall with apache httpClient().
        // which takes atleast 1sec to complete.
        return "Http response as string";
    }).block();
    logger.info("total time needed {}", (System.currentTimeMillis()-time));

    return CompletableFuture.completedFuture(block);

Using Mono.ToFuture() method:

    long time = System.currentTimeMillis();
    CompletableFuture<String> toFuture = Mono.fromCallable(() -> {
        logger.debug("inside in fromCallable() block()");
        //Upstream httpcall with apache httpClient().
        // which takes atleast 1sec to complete.
        return "Http response as string";
    }).toFuture();
    logger.info("total time needed {}", (System.currentTimeMillis()-time));
    return toFuture;

these two code snippets behaves exactly same.

2 Answers

Yes, your doubt is absolutely correct. Actually, Mono.block() and Mono.toFuture() subscribes immediately and brings you out of the reactive system.

This official Spring blog post will make it more clear for you.

I would also recommend going through the source code of Mono which shows block & toFuture subscribing immediately.

Explanation to @ruhul's first comment

The line where your put .toFuture() after your sequence makes the code blocking.

So, in the code below

Mono.fromCallable(() -> {
        logger.debug("inside in fromCallable() block()");
        //Upstream httpcall with apache httpClient().
        // which takes atleast 1sec to complete.
        return "Http response as string";

    }).toFuture(); // this line is the blocking code.

as soon as you confront toFuture(), the subscription of the sequence begins and your code comes out of the reactive context.

-- EDIT: I was wrong. mono.toFuture() doesn't block --

mono.toFuture() isn't blocking. Look at this test:

    @Test
    void testMonoToFuture() throws ExecutionException, InterruptedException {
        System.out.println(LocalTime.now() + ": start");
        Mono<String> mono = Mono.just("hello StackOverflow")
            .delayElement(Duration.ofMillis(500))
            .doOnNext((s) -> System.out.println(LocalTime.now() + ": mono completed"));
        Future<String> future = mono.toFuture();
        System.out.println(LocalTime.now() + ": future created");
        String result = future.get();
        System.out.println(LocalTime.now() + ": future completed");
        assertThat(result).isEqualTo("hello StackOverflow");
    }

This is the result:

20:18:49.557: start
20:18:49.575: future created
20:18:50.088: mono completed
20:18:50.088: future completed

The future is created almost immediately. Half a second later, the mono completes and immediately after that, the future completes. This is exactly what I would expect to happen.

So why does the mono seem blocking in the example provided in the question? It's because of the way mono.fromCallable() works. When and where does that callable actually run? mono.fromCallable() doesn't spawn an extra thread to do the work. From my tests it seems that the callable runs when you first call subscribe() or block() or something similar on the mono, and it will run in the thread that does that.

Here is a test that shows that if you create a mono with fromCallable(), subscribe will cause the callable to be executed in the main thread and even the subscribe() method will seem blocking.

    @Test
    void testMonoToFuture() throws ExecutionException, InterruptedException {
        System.out.println(LocalTime.now() + ": start");
        System.out.println("main thread: " + Thread.currentThread().getName());
        Mono<String> mono = Mono.fromCallable(() -> {
                System.out.println("callabel running in thread: " + Thread.currentThread().getName());
            Thread.sleep(1000);
            return "Hello StackOverflow";
            })
            .doOnNext((s) -> System.out.println(LocalTime.now() + ": mono completed"));
        System.out.println("before subscribe");
        mono.subscribe(System.out::println);
        System.out.println(LocalTime.now() + ": after subscribe");
    }

result:

20:53:37.071: start
main thread: main
before subscribe
callabel running in thread: main
20:53:38.099: mono completed
Hello StackOverflow
20:53:38.100: after subscribe

Conclusion: mono.toFuture() isn't any more blocking than mono.subscribe(). If you want to execute some piece of code asynchronously, you shouldn't be using Mono.fromCallable(). You could consider using Executors.newSingleThreadExecutor().submit(someCallable)

For reference, here is my original (wrong) answer where I belittle the mono.block() method that was assuredly written by people who know a lot more about Java and coding than I do. A personal lesson in humility, I guess.

EVERYTHING BELOW THIS IS NONSENSE

I wanted to verify exactly how this works so I wrote some tests. Unfortunately, it turns out that mono.toFuture() is indeed blocking and the result is evaluated synchronously. I honestly don't know why you would ever use this feature. The whole point of a Future is to hold the result of an asynchronous evaluation.

Here is my test:

@Test
  void testMonoToFuture() throws ExecutionException, InterruptedException {
    Mono<Integer> mono = Mono.fromCallable(() -> {
      System.out.println("start mono");
      Thread.sleep(1000);
      System.out.println("mono completed");
      return 0;
    });
    Future<Integer> future = mono.toFuture();
    System.out.println("future created");
    future.get();
    System.out.println("future completed");
  }

Result:

start mono
mono completed
future created
future completed

Here is an implementation of monoToFuture() that works the way that I would expect it to:

@Test
  void testMonoToFuture() throws ExecutionException, InterruptedException {
    Mono<Integer> mono = Mono.fromCallable(() -> {
      System.out.println("start mono");
      Thread.sleep(1000);
      System.out.println("mono completed");
      return 0;
    });
    Future<Integer> future = monoToFuture(mono, Executors.newSingleThreadExecutor());
    System.out.println("future created");
    future.get();
    System.out.println("future completed");
  }

  private <T> Future<T> monoToFuture(Mono<T> mono, ExecutorService executorService){
    return executorService.submit((Callable<T>) mono::block);
  }

Result:

future created
start mono
mono completed
future completed
Related