在RxJava中，如何在Observable之外反映/提取故障

Question

我们有一个StoreService，它调用一个update（key，content）方法，该方法使用couchbase客户端来执行get->change_content->代替

作为该过程的一部分，我们正在使用Observable retryWhen来处理异常。如果重试次数超过了最大重试次数，它只会传递Exception，然后会触发观测者的onError方法。

在错误无法处理的情况下，我们想做的是从update（key，content）方法向调用它的StoreService抛出一个Exception，但我们未能做到这一点

到目前为止，我们已经尝试了以下方法，但没有成功：

1. 从onError方法抛出一个Exception，但它不会从Observable中抛出
2. 抛出一个RuntimeException，但它不起作用
3. 使用一个包含布尔isFailed成员的DTO：我们在Observable之外创建DTO，如果出现错误，我们会到达订阅者的onError，在那里我们将DTO的isFailed设置为true。Observable完成后，我们检查DTO是否失败，如果是，我们抛出异常。这也不起作用——onError中发生的更改没有在Observable之外传播（为什么？）

以下是伪代码：

 public void update(String key, ConversationDto updateConversationDto) {
    ObservableExecution observableExecution = new ObservableExecution();
    Observable
            .defer(... get the document from couchbase ...) 
            .map(... handle JSON conversion and update the document ...)
            .flatMap(documentUpdate -> {
                return couchbaseClient.getAsyncBucket().replace(documentUpdate);
            })
            .retryWhen(new RetryWithDelay(3, 200))
            .subscribe(
                    n -> logger.debug("on next update document -> " + n.content()),
                    e -> {
                        //logger.error("failed to insert a document",e);
                        observableExecution.setFailure(e);
                    },
                    () -> logger.debug("on complete update document")
            );
    // this is never true
    if (observableExecution.isFailed()) {
        final Throwable e = observableExecution.getFailure();
        throw new DalRuntimeException(e.getMessage(), e);
    }
}

这是重试当代码：

public Observable<?> call(Observable<? extends Throwable> attempts) {
    return attempts
            .flatMap(new Func1<Throwable, Observable<?>>() {
                @Override
                public Observable<?> call(Throwable errorNotification) {
                    if (++retryCount < maxRetries) {
                        // When this Observable calls onNext, the original
                        // Observable will be retried (i.e. re-subscribed).
                        logger.debug(errorNotification + " retry no. " + retryCount);
                        return Observable.timer(retryDelayMillis,
                                TimeUnit.MILLISECONDS);
                    }
                    // Max retries hit. Just pass the error along.
                    logger.debug(errorNotification + " exceeded max retries " + maxRetries);
                    return Observable.error(errorNotification);
                }
            });
}

非常感谢你的帮助！

Answer 1

订阅以异步方式运行，因此isFailed()检查将始终在e -> setFailure(e)代码运行之前立即运行。

正确的方法是从update()方法返回Observable，并在StoreService中订阅它。这样，当你有兴趣处理成功和失败时，你就会得到通知。

Answer 2

我同意@Ross的观点：概念上Observable应该通过update（）返回。我唯一能建议的简化是使用本地可变变量，而不是ObservableExecutionDTO:

public void update(String key, ConversationDto updateConversationDto) {
    final Throwable[] errorHolder = new Throwable[1];
    Observable
        .defer(... get the document from couchbase ...) 
        .map(... handle JSON conversion and update the document ...)
        .flatMap(documentUpdate -> {
            return couchbaseClient.getAsyncBucket().replace(documentUpdate);
        })
        .retryWhen(new RetryWithDelay(3, 200))
        .subscribe(
                n -> logger.debug("on next update document -> " + n.content()),
                e -> {
                    //logger.error("failed to insert a document",e);
                    errorHolder[0] = e;
                },
                () -> logger.debug("on complete update document")
        );
    if (errorHolder[0] != null) {
        final Throwable e = errorHolder[0];
        throw new DalRuntimeException(e.getMessage(), e);
    }
}