RxJava：仅在给定期间的第一项时发出答案

【问题标题】：RxJava: Only emit if the first item for a given periodRxJava：仅在给定期间的第一项时发出
【发布时间】：2017-12-10 14:58:04
【问题描述】：

我想以这样一种方式转换我的源 Flowable，即只有当它们是指定时间段内的第一项时，事件才会发生。

也就是说，我希望第一个项目通过，然后删除所有后续项目，直到有一段时间，比如 10 秒，没有上游事件到达。

请注意，这两者都不是

debounce：这将发出每个项目，如果它在 10 秒内没有跟随另一个项目 - 但这将强制在第一个项目上延迟 10 秒。我想立即发出第一个项目。
throttleFirst：这将发出第一个项目，然后在第一个项目之后丢弃所有后续项目 10 秒。我想在每个上游项目之后重置阻止期。

我现在已经这样解决了：

source
  .flatMap { Flowable.just(1).concatWith(Flowable.just(-1).delay(10, TimeUnit.SECONDS)) }
  .scan(0, { x, y -> x + y })
  .map { it > 0 }
  .distinctUntilChanged()
  .filter { it }

注意：我不关心 source 中的实际项目，只关心它们是否发生 - 但是，当然，我可以将这些项目与 1 或 -1 一起包装在 Pair 中） .

有没有更简单的方法来使用内置的 RxJava(2) 运算符来实现相同的目标？

【问题讨论】：

听起来像以前的问题：stackoverflow.com/questions/41964731/immediate-debounce-in-rx github.com/akarnokd/… 有一个扩展运算符

标签： rx-java rx-java2

【解决方案1】：

这可能会满足您的需要

source.debounce(item -> Observable.timer(10,TimeUnit.SECONDS))

【讨论】：

【解决方案2】：

可以利用switchMap 一次只订阅一个Flowable 并使用布尔值来检查它是否必须发出的事实：

class ReduceThrottle<T>(val period: Long, val unit: TimeUnit) : FlowableTransformer<T, T> {
    override fun apply(upstream: Flowable<T>): Publisher<T> {
        return Flowable.defer {
            val doEmit = AtomicBoolean(true)

            upstream.switchMap { item ->
                val ret = if (doEmit.compareAndSet(true, false)) {
                    // We haven't emitted in the last 10 seconds, do the emission
                    Flowable.just(item)
                } else {
                    Flowable.empty()
                }

                ret.concatWith(Completable.timer(period, unit).andThen(Completable.fromAction {
                    // Once the timer successfully expires, reset the state
                    doEmit.set(true)
                }).toFlowable())
            }
        }
    }
}

那么就是简单的应用transformer：source.compose(ReduceThrottle(10, TimeUnit.SECONDS))。

【讨论】：