没有取消订阅源的 RxJava 超时

Question

我正在订阅 hot observable，然后对其应用 timeout 运算符，但是我不想在抛出 TimeoutException 时取消订阅，只生成特殊项目（我知道来源最终会发出新的物品）。我怎样才能做到这一点？

我正在尝试将超时与 onErrorReturn 结合使用，但这又会导致 onComplete 调用订阅者。

Answer 1

发布、超时和重试（改编自）：

Observable<Long> source =
    Observable.just(100L, 200L, 500L, 1000L, 5000L, 5500L, 6000L)
    .flatMap(v -> Observable.timer(v, TimeUnit.MILLISECONDS).map(a -> v));

source.publish(co -> 
    co.timeout(750, TimeUnit.MILLISECONDS, 
         Observable.just(-1L)
         .concatWith(Observable.error(new RuntimeException()))
    )
    .retry()
 ).blockingForEach(System.out::println);