代码之家 › 专栏 › 技术社区 › Ofek Regev

rxjava在retrywhen on completeable之后完成

reactivex rx-java2

Ofek Regev · 技术社区 · 6 年前

我在completeable上使用retrywhen运算符,有没有方法告诉它从retry flowable完成? 像这样的东西-

PublishSubject<?> retrySubject = PublishSubject.create();
public void someFunction() {
    someCompletable.retryWhen(new Function<Flowable<Throwable>, Publisher<?>>() {
        @Override
        public Publisher<?> apply(Flowable<Throwable> throwableFlowable) throws Exception {
            return throwableFlowable.flatMap(throwable -> retrySubject.toFlowable(BackpressureStrategy.MISSING));
        }
    }).subscribe();
}

public void ignoreError(){
    retrySubject.onComplete();
}

0 回复 | 直到 6 年前

akarnokd 6 年前

你不能阻止 flatMap 给它一个空的来源。同时,每一个错误都会导致越来越多的观察者订阅导致内存泄漏的主题。

使用 takeUntil 要通过另一个源的帮助停止序列,请执行以下操作:

PublishProcessor<Throwable> stopProcessor = PublishProcessor.create();

source.retryWhen(errors -> 
    errors.takeUntil(
        stopProcessor
    )
    .flatMap(error -> Flowable.timer(1, TimeUnit.SECONDS))
)

stopProcessor.onComplete();

编辑如果要重用同一主题,可以在停止路径上禁止项目:

PublishProcessor<Integer> stopProcessor = PublishProcessor.create();

source.retryWhen(errors -> 
    errors.takeUntil(
        stopProcessor.ignoreElements().toFlowable()
    )
    .flatMap(error -> stopProcessor)
)

// retry
stopProcessor.onNext(1);

// stop
stopProcessor.onComplete();

推荐文章

Gissipi_453 · 在RxJava 2中的可观察对象中发出字符串数组时出现“不兼容类型”错误

6 年前

Damn Vegetables · 当N个事件中的任何一个发生时执行,但使用所有其他事件的最后结果

6 年前

Damn Vegetables · 如何压缩(s1,o1),其中s1只发生一次,o1发生多次?

6 年前

CaHa · 如何在rxjava中等待改装响应

6 年前

noloman amram99 · Android和Room:插入作品,查询不

6 年前

Martin · 可观察到。具有状态的生成器,其中异步更新新状态

6 年前

Victor Dmitrienko · RxJava解Bounce运算符取决于消息

6 年前

idunnololz adsion · 调用dispose()之前是否需要检查isDisposed()?

7 年前

Josh Laird · 轮询第二个URL,直到JSON返回预期参数

7 年前

n0shadow · RxJava:为一个完整表列表指定最大并发性

7 年前