我怎样才能返回一个保证不会抛出OnErrorNotImplementedException的RxJava Observa

Question

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

我已经尝试通过使用onErrorReturn, doOnError和onErrorResumeNext连接到可观察对象-没有任何帮助我的情况下。也许我做错了，但我仍然得到rx.exceptions.OnErrorNotImplementedException，如果一个错误发生在可观察对象。

编辑1:这是一个发出错误的可观察对象的例子，我想在一些中间层处理:

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