我如何 return 保证不抛出 OnErrorNotImplementedException 的 RxJava Observable？

Question

我想在我的应用程序中创建一个模式，其中返回的所有 Observable<T> 对象都有一些默认错误处理，这意味着订阅者可以使用 .subscribe(onNext) 重载而不用担心应用程序崩溃. （通常您必须使用 .subscribe(onNext, onError)）。有什么办法可以做到这一点？

我已经尝试使用 onErrorReturn、doOnError 和 onErrorResumeNext 附加到 Observable - 但它们都没有帮助我。也许我做错了，但如果 Observable 中发生错误，我仍然会得到 rx.exceptions.OnErrorNotImplementedException。

编辑 1：这是发出错误的 Observable 示例，我想在某个中间层处理它：

Observable.create(subscriber -> {
    subscriber.onError(new RuntimeException("Somebody set up us the bomb"));
});

编辑 2：我已尝试使用此代码代表消费者处理错误，但我仍然得到 OnErrorNotImplementedException:

// obs is set by the method illustrated in edit 1
obs = obs.onErrorResumeNext(throwable -> {
    System.out.println("This error is handled by onErrorResumeNext");
    return null;
});
obs = obs.doOnError(throwable -> System.out.println("A second attempt at handling it"));
// Consumer code:
obs.subscribe(
    s -> System.out.println("got: " + s)
);

Answer 1

这会起作用 - 关键是 return Observable.empty();

private <T> Observable<T> attachErrorHandler(Observable<T> obs) {
    return obs.onErrorResumeNext(throwable -> {
        System.out.println("Handling error by printint to console: " + throwable);
        return Observable.empty();
    });
}

// Use like this:
Observable<String> unsafeObs = getErrorProducingObservable(); 
Observable<String> safeObservable = attachErrorHandler(unsafeObs);
// This call will now never cause OnErrorNotImplementedException
safeObservable.subscribe(s -> System.out.println("Result: " + s));

我如何 return 保证不抛出 OnErrorNotImplementedException 的 RxJava Observable？

How can I return an RxJava Observable which is guaranteed not to throw OnErrorNotImplementedException?

java

rx-java