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?

Thesaurus

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).
}

Lincolnlincolnshire

onErrorResume returns a Mono with the same type as the stream, so your code wouldn't even compile. You need to move that part to the end. Is there any particular reason why you check the type of the throwable? – Hoem

1. The OP wrote: in case of error i need to return different type of B, let's say, B.buildErrorResponse( a ) ==> B.buildErrorResponse( a ) instance of B , B.buildSuccessResponse( a ) instance of B. (This is what he returned in the question..). 2. I checked the type of the throwable to show to OP how to use onErrorResume. – Lincolnlincolnshire

1 doesn't explain the non-working code. As for 2: There is a variant of onErrorResume that only expects single argument. I see no reason to use another one. – Hoem

I don't see a reason to replacea stream with unchecked exceptions which a new stream. If I get NullpointerException, I would go back to the code. – Lincolnlincolnshire

The onErrorResume part cannot work at this position because it has to return a Mono<A>. Maybe you want to ignore other errors but the OP doesn't. – Hoem

The return type is Mono<B>. – Lincolnlincolnshire

I suggest to consult the documentation. – Hoem

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.

Sideslip

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);
    }

Gerrilee