我们有一个StoreService,它调用一个update(key,content)方法,该方法正在使用couchbase客户端进行get-> change_content-> replace。

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

万一无法处理错误,我们想做的就是将update(key,content)方法中的Exception抛出给调用它的StoreService,但我们没有这样做。

到目前为止,我们已经尝试了以下方法,但均未成功:


从onError方法引发异常,但不会从Observable中抛出该异常。
抛出RuntimeException,但效果不佳。
使用其中包含boolean 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);
    }
}


这是retryWhen代码:

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


非常感谢您的帮助!

最佳答案

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

正确的方法是从Observable方法返回update()并在StoreService中进行预订。这样一来,您就可以在处理成功和失败时收到通知。

关于java - 在RxJava中,如何在Observable之外反射(reflect)/提取故障?,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/29212740/

10-13 07:07