Mono<A>出错时如何返回Mono<B>？

Question

假设我有下面的链：

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

但是如果出现错误，我需要返回不同类型的B，比如说B.buildErrorResponse( a )。任何onError / doOnError方法只能返回原始的Mono类型(A)，但不能返回我需要返回的类型(B)。出错时如何返回不同的Mono类型？

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将捕获该错误信号，并将该流(上游)的所有替换为一个新的流：Mono.just(B.buildErrorResponse(throwable))。

关于someMethod逻辑的附注：

1. 在发生错误的情况下，此方法将发出next信号，该信号表示error信号：B.buildErrorResponse(throwable)。这听起来像是一个坏主意，因为使用someMethod的人必须了解他从upstream (someMthod)收到的next信号是next还是if-condition发出的error信号。相反，让someMethod返回实际的error信号，用户将通过识别例如onErrorResume中的错误来对该错误做出他想要的反应。我对B::buildSuccessResponse.
2. The也有同样的看法，下面的代码可以替换：adapter1.doSomething(arg).

• ：Mono.just(arg).flatMap(adapter1::doSomething)

总而言之，最终的代码将被重构为：

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

Answer 2

您可能想要检查的方法是: onErrorResume(Throwable，Mapper)或onErrorResume(Predicate)

例如，您的代码可能如下所示：

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

在这种情况下，onErrorResume将处理从adapter1::doSomething发出的任何错误。请记住，当一个错误被发出时-没有进一步的'map'，'flatMap‘或任何其他方法在后续的Monos上被调用，因为一个错误将被传递下来，而不是预期的对象。

然后，您可以简单地将所有内容链接起来，如下所示：

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

请记住，所有以“do”开头的方法，如“doOnError”，都将执行所提供的闭包，而不修改“emitted”对象。

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
}

下面是一个简单的测试，您可以使用它来测试行为

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