当嵌套错误处理程序调用 onExceptionResumeNext 时，Observable 停止发射

Question

在下面的代码中，我有一个嵌套的 observable。 flatMap 中的 sendMessage 调用 sendMessage 函数，该函数也是一个 observable。如果在这个嵌套的 observable 中发生异常，则假定 onExceptionResumeNext 会捕获异常，处理异常然后继续，就好像什么都没发生一样。异常确实被捕获，但是一旦对异常的处理完成，就不会在流中进行进一步的排放。甚至没有调用 doOnComplete。本质上，onExceptionResume next 只是挂起。

我尝试过 onErrorReturnItem 但结果相同。我没有在 Stackoverflow 或其他地方找到一个例子，它甚至在嵌套的 observable 中显示 onExceptionResumeNext 或 onErrorResumeNext 或 onErrorReturnItem ，经过一天的工作，我怀疑它可能无法支持嵌套的错误处理程序。

注意：在 onExceptionResumeNext 我目前只是返回

Observable.empty<MessageToSend>()

在我的实际代码中，我有处理异常的代码，我尝试返回一个 observable 以及只返回数据。不管我做什么 - 它总是挂起。

fun postMessages() {
    val msgToSendPublisher = BehaviorSubject.createDefault(MessageToSend())

    msgToSendPublisher
        .flatMap { _ ->
            App.context.repository.getMessageToSend().toObservable()
        }
        .doOnError { error ->
            if (error is EmptyResultSetException)
                App.context.repository.setSendStatusToNotSendingForAllMessages()
        }
        .doOnNext { messageToSend ->
            App.context.repository.updateMessage(messageToSend)
        }
        .flatMap { messageToSend ->
            App.context.repository.sendMessage(messageToSend)
        }
        .doOnNext { messageToSend ->
            messageToSend.dateSent = Date()
            App.context.repository.updateDateLastMessageSent(messageToSend)
        }
        .doOnNext { messageToSend ->
            if (messageToSend.totalMessagesToSend == 1)
                App.context.repository.updateSendStatus(messageToSend, MessageSendStates.NOT_SENDING)
            else
                Observable.just(messageToSend)
        }
        .doOnNext {
            msgToSendPublisher.onNext(it)
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { messageToSend ->

            },
            { ex ->
                onMessagesSent()
            },
            {
                onMessagesSent()
            }
        )
}


fun sendMessage(messageToSend: MessageToSend): Observable<MessageToSend> {
    val obs = Observable.fromCallable {
            if (messageToSend.totalMessagesToSend == 3)
                throw Exception("Couldn't send to recipient.")

            messageToSend
        }.map {
            storeMessageSent(messageToSend)
        }.onExceptionResumeNext {
            Observable.empty<MessageToSend>() // Hangs here.
        ).doOnComplete {
            addNewMessageIfRequired(messageToSend, newMessage)
        }

        return obs
}

更新：

我决定测试我发现的使用 onExceptionResumeNext 的示例代码。它看起来像这样：

    Observable.fromArray(1, 2, 3)
        .doOnNext {
            if (it == 2) {
                throw (RuntimeException("Exception on 2"))
            }
        }
        .onExceptionResumeNext(
            Observable.just(10)
        )
        .subscribe(
            {
                var x = it
            },
            {
                var x = it
            },
            {
                var x = 0
                x++
            }
        )

如果您在 onExceptionResumeNext 内的行上放置一个断点，它将在您第一次运行 observable 时被调用，而不仅仅是在抛出异常时。这显然是 RxJava 文档中未标识的行为。任何开发人员都会认为只有在抛出异常时才会调用它。在上面的示例中，将值设置为 10 并不是真正的问题。它实际上只是为发生异常的情况设置返回值。但是，如果这是在数据库中存储内容的更精细的代码（我的应用程序会这样做），那么它会在可观察对象初始化时被调用——这真的很糟糕。尽管有这个发现，它仍然没有解决我的问题，因为没有发出更多的项目。我在示例代码中发现的是，当调用 onExceptionResumeNext 时，也会调用 onComplete。太糟糕了，文档也没有提到。

Answer 1

您可能希望使用defer 来推迟函数调用的执行，这会在调用时产生副作用：

Observable<Integer> createFallback() {
    System.out.println("Why is this executing now?!");
    return Observable.empty();
}

Observable.<Integer>error(new Exception())
.onExceptionResumeNext(createFallback())
.subscribe();

createFallback 运行是因为您指定它通过调用它来运行。如果序列被改写，原因应该会更明显：

Observable<Integer> fallback = createFallback();

Observable.<Integer>error(new Exception())
.onExceptionResumeNext(fallback)
.subscribe();

现在如果你注释掉error-observable部分，它仍然执行createFallback()吗？是的，RxJava 甚至还没有参与到这一点上。

如果您希望这种副作用不会发生在 createFallback 上，则必须推迟整个方法的执行，为此有一个运算符：defer：

Observable.<Integer>error(new Exception())
.onExceptionResumeNext(Observable.defer(() -> createFallback()))
.subscribe();

我认为这在 Kotlin 中看起来像这样：

Observable.error(new Exception())
.onExceptionResumeNext(Observable.defer { createFallback() })
.subscribe()