RxJava onErrorResumeNext 调度器答案

【问题标题】：RxJava onErrorResumeNext schedulerRxJava onErrorResumeNext 调度器
【发布时间】：2016-05-05 22:20:50
【问题描述】：

我有一个 observable，它可能会因特殊异常而失败，在这种情况下，我想显示一个带有重试按钮的对话框。我看过this 的回答，但它并没有完全达到我想要的效果。我无法使用retryWhen 来解决我的问题，所以我使用了onErrorResumeNext。如果你能想出一个方法来用retryWhen做同样的事情，请告诉。

现在我有这段代码：

public Observable<Order> proceedWithOrdering(Activity activity) {
    return apiService.createOrder()
            .subscribeOn(Schedulers.io())
            .compose(applyRetryLogic(activity))
            .subscribeOn(AndroidSchedulers.mainThread())
            .observeOn(AndroidSchedulers.mainThread());
}

public <T extends ApiResponse> Observable.Transformer<T, T> applyRetryLogic(Activity activity) {
    return observable -> observable
            .onErrorResumeNext(retry(observable, activity))
            .subscribeOn(AndroidSchedulers.mainThread());
}

public <T> Func1<Throwable, ? extends Observable<? extends T>> retry(Observable toRetry, Activity activity) {
    return throwable -> {
        if (throwable instanceof NetworkException) {
            MaterialDialog dialog = retryDialog(activity);
            View retry = dialog.getActionButton(DialogAction.POSITIVE);
            View cancel = dialog.getActionButton(DialogAction.NEGATIVE);
            Observable<Object> retryClick = RxView.clicks(retry).map(o -> {
                dialog.dismiss();
                return o;
            });
            Observable<Object> cancelClick = RxView.clicks(cancel).flatMap(o -> {
                dialog.dismiss();
                return Observable.error(throwable);
            });

            dialog.show();

            return Observable.amb(retryClick, cancelClick)
                    .flatMap(o -> toRetry.compose(applyRetryLogic(activity)));
        } else {
            return Observable.error(throwable);
        }
    };
}

问题是retry 中的call 不在主线程上执行，它引发了Can't create handler inside thread that has not called Looper.prepare() 异常。

问题是 - 我如何强制它在主线程上执行？如您所见，我已经尝试在compose 和onErrorResumeNext 之后执行subscribeOn(AndroidSchedulers.mainThread())，但没有成功。

我已经使用不在单独线程上运行的简单可观察对象测试了我的代码，并且运行良好。

【问题讨论】：

标签： java android rx-java

【解决方案1】：

您可以通过flatMapping PublishSubject 来完成此操作，然后在按下相关按钮后更新。这是一个经典的 Java Swing 示例。

public class RetryWhenEnter {
    public static void main(String[] args) {
        AtomicInteger d = new AtomicInteger();
        Observable<Integer> source = Observable.just(1);

        source.flatMap(v -> {
            if (d.incrementAndGet() < 3) {
                return Observable.error(new RuntimeException());
            }
            return Observable.just(v);
        })
        .retryWhen(err -> {
            return err.flatMap(e -> {
                System.out.println(Thread.currentThread() + " Error!");
                PublishSubject<Integer> choice = PublishSubject.create();
                SwingUtilities.invokeLater(() -> {
                    int c = JOptionPane.showConfirmDialog(null, 
                        e.toString() + "\r\nRetry?", "Error",
                        JOptionPane.YES_NO_OPTION);
                    if (c == JOptionPane.YES_OPTION) {
                        choice.onNext(1);
                    } else {
                        choice.onCompleted();
                    }
                });
                return choice;
            });
        }).subscribe(System.out::println, 
                Throwable::printStackTrace);
    }
}

编辑：

或在onErrorResumeNext 之前使用observeOn(AndroidSchedulers.mainThread()) 或在使用retryWhen 时：retryWhen(o -> o.observeOn(AndroidSchedulers.mainThread())...)。

编辑 2 我已回滚更改，因此答案再次有意义。

【讨论】：

我比retryWhen 更喜欢第二种解决方案，谢谢。我已针对 Android 调整了您的第一个解决方案，并将其作为单独的答案发布，供其他用户查看。
这retryWhen(o -> o.observeOn(AndroidSchedulers.mainThread())...) 帮助了我。点赞。

【解决方案2】：

有一种方法可以使用retryWhen 解决我的问题（感谢@akarnokd）：

public <T extends ApiResponse> Observable.Transformer<T, T> applyRetryLogic(Activity activity) {
    return observable -> observable
            .retryWhen(err -> err.flatMap(throwable -> {
                L.d(Thread.currentThread() + " Error!");
                if (throwable instanceof NetworkException) {
                    PublishSubject<Integer> choice = PublishSubject.create();
                    activity.runOnUiThread(() -> {
                        MaterialDialog dialog = retryDialog(activity);
                        View retry = dialog.getActionButton(DialogAction.POSITIVE);
                        View cancel = dialog.getActionButton(DialogAction.NEGATIVE);
                        RxView.clicks(retry).subscribe(o -> {
                            dialog.dismiss();
                            choice.onNext(1);
                        });
                        RxView.clicks(cancel).subscribe(o -> {
                            dialog.dismiss();
                            choice.onError(throwable);
                        });

                        dialog.show();
                    });
                    return choice;
                } else {
                    return Observable.error(throwable);
                }
            }));
}

【讨论】：