How to use Mono's content in Reactor

Question

I'm trying to call a reactive REST API to fetch deadlinesTS variable. Then I'm trying to set the same in my Pojo class. But the value after setting the deadlineTS in BOLCompliance is not consistent using subscribe(). Sometimes I'm able to set the value, other times I get null. How can I make sure everytime I'm able to set the value.

Mono<String> deadlineTS = portCallServiceCaller.getDeadlineTSByComplianceId(compliance.getId());

BOLCompliance complianceResponse = new BOLCompliance();
deadlineTS.subscribe(val->complianceResponse.setDeadlineTimestamp(val));

Answer 1

Use Mono's content outside a reactive pipeline (blocking)

You can use block() method like this:

Mono<String> nameMono = Mono.just("some-value").delayElement(Duration.ofMillis(300));

Person person = new Person();
person.setName(nameMono.block());

System.out.println(person.getName());

This triggers the operation and waits for its completion. Note that the calling thread blocks.

Alternatively, you could use subscribe(consumer, errorConsumer, completeConsumer) and provide a Runnable that will be triggered when the operation is completed:

valueMono.subscribe(v-> person.setName(v), throwable -> {}, () -> System.out.println(person.getName()));

However, the subscribe() method will return immediately.

Use Mono's content within a reactive pipeline

You can choose one of the operators provided depending on the case.

In this case, you can use map operator to transform the String into BOLCompliance:

  Mono<BOLCompliance> fetchBOLCompliance() {
    Mono<String> deadlineMono = portCallServiceCaller.getDeadlineTSByComplianceId(compliance.getId();

    return deadlineMono.map(deadline -> {
      BOLCompliance compliance = new BOLCompliance();
      compliance.setDeadlineTimestamp(deadline);
      return compliance;
    });
  }

If you would like to run an asynchronous task (eg. database access) you would need to use flatmap operator.

subscribe() returns immediately

According to Javadoc:

Disposable subscribe(Consumer<? super T> consumer)

Keep in mind that since the sequence can be asynchronous, this will immediately return control to the calling thread. This can give the impression the consumer is not invoked when executing in a main thread or a unit test for instance.

In other words, the subscribe method kicks odd the work and returns immediately. So you get no guarantee that the operation is done. For instance, the following example will always end up with a null value:

Mono<String> nameMono = Mono.just("some-value").delayElement(Duration.ofMillis(300));

Person person = new Person();
nameMono.subscribe(v-> person.setName(v));

System.out.println(person.getName());

Here, the person.getName() method is invoked immediately while person.setName(v) is invoked after 300 milliseconds.