java - RxJava 2 - 如何在错误时取消无限流并处理它？

Question

我有以下无限流，每秒都会执行一些操作。
我想要的是在出现错误时停止流并处理它。
我怎样才能做到这一点？

void doSomething() {
        Disposable disposable = execute(doSomethingInner(), 0L, TimeUnit.SECONDS, schedulerProvider.io(), someClass -> 1L).doOnError
                (throwable -> {
            Timber.e(throwable, "error happened");// Never triggered
        })
                .doOnNext(someClass -> Timber.i("doing the infinite stuff"))
                .subscribe(Functions.emptyConsumer(), throwable -> {
                    Timber.e(throwable, "stop doing the infinite stuff");// Never triggered
                });
    }

    Observable<SomeClass> doSomethingInner() {
        return Observable.error(new Exception("something went wrong"));
    }

    Observable<SomeClass> execute(Observable<SomeClass> source,
                                  long delayInterval,
                                  TimeUnit timeUnit,
                                  Scheduler scheduler,
                                  Function<SomeClass, Long> interval) {
        return Observable.defer(new Callable<ObservableSource<SomeClass>>() {
            long currentInterval = delayInterval;

            @Override
            public ObservableSource<SomeClass> call() {
                return Single.timer(currentInterval, timeUnit, scheduler)
                        .flatMapObservable(o -> source)
                        .doOnNext(t -> currentInterval = interval.apply(t));
            }
        })
                .repeat()
                .retry();
    }

Answer 1

我认为retry()正在消耗你的错误。
尝试以下任一操作:

• 删除此 retry()完全
• 或将其更改为 retry(Predicate<Throwable>)决定是否重复。

如果您不提前使用流，订阅者的默认行为是在错误时取消流，并且您应该收到 onError() 的回调。里面subscribe() .