RxJava2 materialize() 出错后不发出下一项

Question

我一直在尝试将我的 onErrors 转换为通知，以保持流发射项目。据我了解，materialize() 运算符就是这样做的。所以基本上：

materialize() / dematerialize() 可用于开启终端事件 进入通知

所以我根据这个问题（How to continue streaming items after error in RxJava?）对此进行了测试。我尝试了以下方法：

 @Test
public void materializeTest() {
    final Observable<String> stringObservable = Observable.fromArray("1", "2", "3")
            .flatMap(x -> {
                if (x.equals("2")) {
                    return Observable.error(new NullPointerException());
                }

                return Observable.just(x);
            })
            .materialize()
            .map(n -> n.getValue());

    final TestObserver<String> testObs = stringObservable.test();
    Java6Assertions.assertThat(testObs.values().size()).isEqualTo(2);

    testObs.assertValueAt(0, "1");
    testObs.assertValueAt(1, "3");
}

结果是在“2”给出错误后不再发出任何项目。我还尝试对自己的 Notification 对象 (MyNotification<T>) 进行变形并执行以下操作：

stringObs
  .map(string -> MyNotification.success(string)
  .onErrorReturn(error -> MyNotification.error())

但最终结果始终相同：在“2”之后不再发出任何项目。我 100% 做错了什么，但我无法真正理解是什么。

Answer 1

对于flatMap，如果内部Observables 之一失败，则序列终止，不再从上游转换任何项目。这发生在之前 materialize() 甚至参与其中。

因此，与其尝试实现合并的流程，不如单独实现内部源：

Observable.fromArray("1", "2", "3")
        .flatMap(x -> {
            if (x.equals("2")) {
                return Observable.<String>error(new NullPointerException())
                                 .materialize();
            }

            return Observable.just(x)
                             .materialize();
        })
        .filter(n -> n.isOnNext())
        .map(n -> n.getValue());