rx-java - 如何避免 Rx 流中的重复网络调用？

Question

我有以下流。

Observable.just(Unit) // execute immediately
    .mergeWith(tryAgainRelay) // execute again when this relay emits
    .flatMap {
      api.call() // emits sealed class single containing `Success` or `Error` state
          .toObservable()
          .startWith(Loading) // emit loading (same sealed class as above)
    }

但是，我从不想要重复(同时)的网络调用(内部平面图)。如果 tryAgainRelay 在网络调用已经在进行中时发射，我希望丢弃该发射。我可以使用以下代码避免它们:

var isLoading = false // track whether api call is in flight
Observable.just(Unit)
    .mergeWith(tryAgainRelay)
    .filter { isLoading.not() } // prevent emissions if api call is already in flight
    .flatMap {
      api.call()
          .toObservable()
          .startWith(Loading)
          // update state variable
          .doOnNext {
            isLoading = when (it) {
              Loading -> true
              is Error, is Success -> false
            }
          }
        }

它有点难看，涉及使用 .doOnNext() 步出流。有没有更好、更惯用的方法来做到这一点？

Answer 1

使用 concatMap 而不是 flatMap。

重试将依次“排队”创建，因此您可以确保只有一个请求将并行执行(下一个请求将在前一个请求完成时开始 - 完成或错误 - 和 如果 tryAgainRelay 已发出)。