How to return Mono<B> from Mono<A> on error?

Question

Let's say i have the following chain:

public Mono<B> someMethod( Object arg ) {
  Mono<A> monoA = Mono.just( arg ).flatMap( adapter1::doSomething );
  // success chain
  return monoA.map( B::buildSuccessResponse );
}

But in case of error i need to return different type of B, let's say, B.buildErrorResponse( a ). Any onError / doOnError methods can return only original Mono type (A), but not the one I need to return (B). How to return different Mono type in case of error?

Answer 1

public Mono<B> someMethod(Object arg) {
    return Mono.just(arg)
            .flatMap(adapter1::doSomething)
            .onErrorResume(throwable -> throwable instanceof Exception,
                           throwable -> Mono.just(B.buildErrorResponse(throwable)))
            .map(B::buildSuccessResponse);
}

onErrorResume will catch the error signal if it was thrown in upstream and will replace all that stream (the upstream) with a new stream: Mono.just(B.buildErrorResponse(throwable)).

---

Side note about someMethod logic:

1. In case of an error, this method will signal a next signal which indicates error signal: B.buildErrorResponse(throwable). It sounds like a bad idea because who will use someMethod must understand if the next signal he received from upstream (someMthod) is next or error signal by if-condition. Instead, let someMethod return the actual error signal and the user will react to that error as he wants by recognizing the error in, for example, onErrorResume.

2. I have the same argument for B::buildSuccessResponse.

3. The following code can be replaced: Mono.just(arg).flatMap(adapter1::doSomething) with adapter1.doSomething(arg).

All in all, the final code will be refactored to:

public Mono<B> someMethod(Object arg) {
    return adapter1.doSomething(arg).
}

Answer 2

The methods you might want to check out are: onErrorResume(Throwable, Mapper) or onErrorResume(Predicate)

For example your code could look something like this:

public Mono<B> someMethod( Object arg ) {
  Mono monoA = Mono.just( arg ).flatMap( adapter1::doSomething );
  return monoA.onErrorResume({ Throwable e ->
      return B::buildSuccessResponse
  });
}

In this case the onErrorResume would handle any error emitted from adapter1::doSomething. Keep in mind that when an error is emitted - no further 'map', 'flatMap' or any other method is invoked on subsequent Monos since an error will be passed down instead of the expected object.

You could then simply chain it all to look like:

public Mono<B> someMethod( Object arg ) {
  return Mono.just( arg ).flatMap( adapter1::doSomething ).onErrorResume({ Throwable e ->
      return B::buildSuccessResponse
  });
}

Keep in mind that all method that start with 'do' like 'doOnError' will execute the closure provided without modifying the 'emitted' object.

Answer 3

public Mono<B> someMethod( Object arg ) {
  Mono<A> monoA = Mono.just( arg ).flatMap( adapter1::doSomething );

  return monoA
   .map( B::buildSuccessResponse )
   .onErrorResume({ Throwable e ->
      return  Mono.just(B.buildSuccessResponse(e));
   }); // if there is no error, `onErrorResume` is effectively doing nothing
}

Below is a simple test which you can use to test the behaviour

    public static void main(String[] args) {
        Mono.just(1)
                .doOnNext(x -> {
                    throw new RuntimeException("some error from previous operation");
                }) //comment this out if you want to simulate successful response
                .map(String::valueOf)
                .onErrorResume(e -> Mono.just("error response"))
                .subscribe(System.out::println);
    }