RxJava mergeDelayError OnErrorFailedException Fehler: trat auf, wenn Fehler zu propagieren versucht Observer.onError

Question

Wenn ich mergeDelayError auf einer Liste von Observablen verwenden, um jede beobachtbare auszuführen, bevor erroring aus, erhalte ich:

OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError 

Code:

return Observable.mergeDelayError(Observable.from(observableList)); 

Wo ich die Observable mit einem Argument für onError abonnieren (die auch nicht ausgeführt wird).

Wenn ich jedoch versuchte, onErrorResumeNext zu verwenden, um dies zu debuggen, fand ich, dass das Verbreiten des beobachtbaren Fehlers keine Probleme verursachte. Was passiert hier?

return Observable.mergeDelayError(Observable.from(observableList)).onErrorResumeNext(new Func1<Throwable, Observable<? extends Pair<String, Map<Long, Set<ProgressGradedItemBL>>>>>() { 
    @Override 
    public Observable<? extends Pair<String, Map<Long, Set<ProgressGradedItemBL>>>> call(Throwable throwable) { 
     return Observable.error(throwable); 
    } 
}); 

Hier ist der Stacktrace ich ohne onErrorResumeNext erhalten:

rx.exceptions.OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError 
    at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:175) 
    at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:97) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:466) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:599) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542) 
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceProducer.tick(OnSubscribeCombineLatest.java:151) 
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceProducer.onCompleted(OnSubscribeCombineLatest.java:183) 
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceRequestableSubscriber.onCompleted(OnSubscribeCombineLatest.java:236) 
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:435) 
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:435) 
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472) 
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614) 
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542) 
    at rx.internal.operators.OperatorDistinctUntilChanged$1.onCompleted(OperatorDistinctUntilChanged.java:64) 
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53) 
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43) 
    at rx.internal.operators.OperatorFilter$1.onCompleted(OperatorFilter.java:42) 
    at rx.internal.operators.OperatorSubscribeOn$1$1$1.onCompleted(OperatorSubscribeOn.java:66) 
    at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:99) 
    at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:46) 
    at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:158) 
    at rx.internal.operators.OperatorConcat$ConcatSubscriber.completeInner(OperatorConcat.java:150) 
    at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onCompleted(OperatorConcat.java:212) 
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53) 
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53) 
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43) 
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53) 

Answer 1

Die OnErrorFailedException wird ausgelöst, wenn Sie die subscribe(Action1) statt subscribe(Action1, Action1) verwenden. Außerdem scheinen Sie eine ältere Version von RxJava zu verwenden (der Name drainAndComplete war wie 1.0.12 und vorher). Die aktuelle Version ist 1.1.2.

Answer 2

Wenn Ihr Stream eine ungeprüfte Ausnahme gegen das Observable wirft, wird die beobachtbare Verarbeitung abgebrochen und eine OnErrorFailedException ausgelöst.