How to design/create reactive method with Project Reactor

user_8885367 0
Viewed 4386

I am newly on project-reactor based reactive framework like spring-webflux and I have a few question about it.

Question 1:

Example 1:

  public Mono<String> doSome(String str){
    String key = str.substring(0, 5).toLowerCase() + str.substring(6);
    return getValueFromRedis(key);
  }

  public Mono<String> getValueFromRedis(String key){
    return reactiveRedisOperations().opsForHash().get("TestCache", key);
  }

Example 2:

  public Mono<String> doSome(String str){
    return Mono.fromCallable(() -> {
      String key = str.substring(0, 5).toLowerCase() + str.substring(6);
      return getValueFromRedis(key);
    }).flatMap(stringMono -> stringMono);
  }

  public Mono<String> getValueFromRedis(String key){
    return reactiveRedisOperations().opsForHash().get("TestCache", key);
  }

is there a difference between with two example or both of them are acceptable.

Question 2:

Example 1: 

  @PostMapping(value = "/greet")
  public Mono<String> greet(String name) {
    return Mono.fromCallable(() -> aMethod(name));
    // or return Mono.just(aMethod(name));
  }

  public String aMethod(String name){
    return "Hello: " + name;
  }

Example 2: 

  @PostMapping(value = "/greet")
  public Mono<String> greet(String name) {
    return aMethod(name);
  }

  public Mono<String> aMethod(String name){
    return Mono.just("Hello: " + name);
  }

Second question is weird I know but I'm wonder all of method should return Mono or Flux or can I use like Question2/Example1.

project-reactor spring-boot java
2 Answers

Question1: yes, there is a difference. In Example 1 you are creating String key outside of Mono.fromCallable. It's not a big deal here, but if it was expensive operation you would slow everything down.

Also this logic reactiveRedisOperations().opsForHash() is executed out of Mono.fromCallable. Same thing - if this was expensive, you are slowing everything down.

Question2: same point as in question 1. Mono.just accepts a regular object, not something that will be called later when needed (like Callable or Supplier). So when using Mono.just you are paying the price of argument initialization immediately.

In your examples there is barely any difference, but Mono and Flux are usually used to chain expensive operation in asynchronous way, so nothing is blocked, like database calls or calls to other external services. Look at my example below where sleep is a simulation of external call. Notice the order of print statements.

public String superLongMethod() {
  System.out.println("superLongMethod");
  Thread.sleep(10000);
  return "ok";
}

System.out.println("before");
Mono.just(superLongMethod());
System.out.println("after");

// printed output is - before - superLongMethod - after

-----------------------------------------------------------------

System.out.println("before");
Mono.fromCallable(() -> superLongMethod());
System.out.println("after");

// printed output is - before - after - superLongMethod

-----------------------------------------------------------------

System.out.println("before");
String key = superLongMethod();
System.out.println("after");
return getValueFromRedis(key);

// printed output is - before - superLongMethod - after

-----------------------------------------------------------------

System.out.println("before");
Mono<String> mono = Mono.fromCallable(() -> {
  String key = superLongMethod();
  return getValueFromRedis(key);
}).flatMap(stringMono -> stringMono);
System.out.println("after");
return mono;

// printed output is - before - after - superLongMethod
user_48966510

fromCallable executes lambda every time Mono has a new subscriber whereas just captures its argument at the time of instantiation and emits it to each of its subscribers.

Following code illustrates the difference:

private Instant getTime() {
    final Instant now = Instant.now();
    System.out.println("getTime(): " + now);
    return now;
}

@Test
public void just() throws InterruptedException {
    final Mono<Instant> mono = Mono.just(getTime())
            .doOnNext(instant -> System.out.println(instant));

    Thread.sleep(500);

    mono.subscribe();

    Thread.sleep(500);

    mono.subscribe();

    /* output is
        getTime(): 2019-08-14T22:47:06.823Z
        2019-08-14T22:47:06.823Z
        2019-08-14T22:47:06.823Z
    */
}

@Test
public void fromCallable() throws InterruptedException {
    final Mono<Instant> mono = Mono.fromCallable(() -> getTime())
            .doOnNext(instant -> System.out.println(instant));

    Thread.sleep(500);

    mono.subscribe();

    Thread.sleep(500);

    mono.subscribe();

    /* output is
        getTime(): 2019-08-14T22:47:13.947Z
        2019-08-14T22:47:13.947Z
        getTime(): 2019-08-14T22:47:14.447Z
        2019-08-14T22:47:14.447Z
    */
}

(Also you may have a look at RxJava Single.just() vs Single.fromCallable()? which is about RxJava but its API is almost identical to the Reactor).

So the answer to your Question 1: yes, there is a difference. In Example 1 invocation of doSome will immediately lead to the asynchronous value fetching from Redis. In Example 2 Redis will not be involved until someone subscribes to the Mono returned by the doSome method.

user_3554380
Related