rx-java - 可观察序列终止时的异步清理

Question

我需要在可观察到的完成/失败时执行异步方法(例如清理作业)。此外，如果清理方法失败，可观察链也应该失败。

有没有标准的方法来做到这一点？

假设我有一个序列 Observable 源和返回一个 observable 清理结果的 asyncCleanup() 方法。

doOnCompleted/doOnTerminate/doOnUnsubscribe等副作用方法好像不太合适:

source.doOnUnsubscribe(()->asyncCleanup().subscribe());

即使 asyncCleanup() 失败，可观察链也会成功。所以 asyncCleanup() 应该是同一链的一部分。

我想到的最好的是:

source.onErrorResumeNext(err->asyncCleanup()
         .flatMap(cleanupResult->Observable.error(err)))
      .concatWith(asyncCleanup()
         .flatMap(cleanupResult->Observable.empty()));

如果失败，onErrorResumeNext 将调用 asyncCleanup() 并将映射回原始错误。如果成功，我们可以连接映射到空序列的 asyncCleanup()。不幸的是，如果下游有 take() 或类似的限制运算符，它就不会工作，这可能导致连接的可观察对象甚至不订阅。

2017 年 8 月 1 日更新:这个问题具体是关于序列可观察量的。对于单项可观察源，解决方案非常简单:

singleItemSource
    .onErrorResumeNext(err->asyncCleanup()
        .flatMap(cleanupResult->Observable.error(err)))
    .flatMap(single->asyncCleanup()
        .map(cleanupResult->single));

Answer 1

您提出的想法打破了 Observable 契约，因为:

Rx Design Guidelines, 6.8 - Unsubscription should not throw

我想到的唯一解决问题的方法是将 asyncCleanup() 转换为与前面序列相同的类型，如果成功则返回例如 Observable.empty()，您可以使用 concat 运算符。

public Observable<Long> asyncCleanup() {
...
    return Observable.empty();
}

//Concat asyncCleanup() to be executed after the sequence
Observable.rangeLong(1, 10)
            .concatWith(asyncCleanup())
            .subscribe(System.out::println, System.out::println);

希望对你有帮助