如何在重试之前使超时运算符在 RxJava2（RxJava3）链中触发一次异常

Question

我正在尝试扩展我的自定义FlowableTransformer 的功能，我使用它有条件地从一个Flowable 切换到另一个，由提供的Function<T, Flowable<R>> 返回，即在我的switchFunction 中我将一个项目发送到服务器并切换如果初始项目匹配某些条件，则向服务器响应流动：

/**
 * Switches from upstreamFlowable to anotherFlowable
 * using switchFunction if item from upstreamFlowable
 * matches test condition
 *
 * usage:
 *       upstreamFlowable
 *       .compose(RxSwitchFlowablesOnCondition(condition, switchFunction))
 **/
open class
RxSwitchFlowablesOnCondition<T, R>(private val condition : Predicate<T>,
                                   private val switchFunction : Function<T, Flowable<R>>) : FlowableTransformer<T, R>
{
  override fun
  apply(upstreamFlowable : Flowable<T>) : Publisher<R>
  {
    return upstreamFlowable
    .compose(RxFlowableOnIo<T>())
    .filter {emittedItem : T ->
      condition.test(emittedItem)
    }
    .switchMap {emittedItem : T ->
      val anotherFlowable = switchFunction.apply(emittedItem)

      anotherFlowable
      //.timeout(CONNECTION_TIMEOUT_SECONDS, SECONDS) // #1
    }
    //.timeout(CONNECTION_TIMEOUT_SECONDS, SECONDS) // #2
  }
}

如果发生错误，要重新启动链，在我的 android 应用程序中，我使用 retry 运算符：

upstreamFlowable
.compose(RxSwitchFlowablesOnCondition(condition, switchFunction))
.retry() // to resubscribe on error
.subscribe(...)

我想做的是添加timeout逻辑，例如，如果upstreamFlowable发出一个项目和在anotherFlowable获得响应之间的时间大于CONNECTION_TIMEOUT_SECONDS，我想触发一个错误并重新订阅upstreamFlowable，这样我就可以等待下一个emittedItem。

我有什么：

1. 如果我取消注释行#1，则超时会产生TimeoutException，重试重新订阅链，即使没有下一个发射项目，我也会在每个CONNECTION_TIMEOUT_SECONDS 时间段继续收到超时异常。 如何跳过它们？
2. 如果我取消注释行#2，那么我会在订阅后CONNECTION_TIMEOUT_SECONDS 之后收到TimeoutExceptions，而不是自项目发射后。 如何解决？

需要你的帮助。提前谢谢！

Answer 1

要获得您想要的行为，您只需在 switchMap 块中进行少量更改。所以，我将你的 apply 实现修改为这个

override fun
  apply(upstreamFlowable : Flowable<T>) : Publisher<R>
  {
    return upstreamFlowable
    .compose(RxFlowableOnIo<T>())
    .filter {emittedItem : T ->
      condition.test(emittedItem)
    }
    .switchMap {
        Flowable.just(it)
            .switchMap { emittedItem : T ->
                val anotherFlowable = switchFunction(emittedItem)

                 anotherFlowable
              }
             .timeout(CONNECTION_TIMEOUT_SECONDS, SECONDS)
         }
  }

通过这样做，timeout 将在发出值时开始计数。

更新 为了测试，我已经编写了简单的代码并添加了日志：

1. 上游将在 2 秒后发射一次
2. 另一个Flowable 将在 1 秒后发射一次
3. 超时为 500 毫秒

// 示例用法

val upstream = Flowable.timer(2, TimeUnit.SECONDS)
        upstream
                .doOnSubscribe { println("subscribed ${System.currentTimeMillis()}") }
                .compose(
                        RxSwitchFlowablesOnCondition<Long, Long>(
                                Predicate { true },
                                { Flowable.timer(1, TimeUnit.SECONDS)})
                )
                .doOnError { println("timeout ${System.currentTimeMillis()}") }
                .retry()
                .subscribe()

// RxSwitchFlowablesOnCondition

open class
    RxSwitchFlowablesOnCondition<T, R>(private val condition : Predicate<T>,
                                       private val switchFunction : (T) -> Flowable<R>) : FlowableTransformer<T, R>
    {
        override fun
                apply(upstreamFlowable : Flowable<T>) : Publisher<R>
        {
            return upstreamFlowable
                    .observeOn(Schedulers.io())
                    .filter {emittedItem : T ->
                        condition.test(emittedItem)
                    }
                    .switchMap {
                        Flowable.just(it)
                                .doOnNext { println("onNext ${System.currentTimeMillis()}") }
                                .switchMap { emittedItem : T ->
                                    val anotherFlowable = switchFunction(emittedItem)

                                    anotherFlowable
                                }
                                .timeout(500, TimeUnit.MILLISECONDS)
                    }

        }
    }

这是示例输出日志

subscribed 1569286225769
onNext 1569286227790
timeout 1569286228292  // around 500 ms after onNext
subscribed 1569286228295
onNext 1569286230297
timeout 1569286230800
subscribed 1569286230800
onNext 1569286232803
timeout 1569286233306
subscribed 1569286233306

符合你的预期吗？