使用 RxJava 对状态重复操作答案

【问题标题】：Repeating actions on state with RxJava使用 RxJava 对状态重复操作
【发布时间】：2017-05-31 01:10:15
【问题描述】：

这是在 Kotlin 中，但我认为任何编写 Java 的人都能理解。

我正在尝试使用 Rx 制作秒表，但在执行实际停止和启动时遇到了一些麻烦。最大的问题是我不知道如何保持当前时间，同时随着不同的动作（开始和停止）进入而对其进行修改。这就是我现在所拥有的。

fullTime.switchMap { startTime ->
    controlCommands.switchMap { command ->
        when (command) {
            ControlState.PLAY -> Observable.interval(1L, TimeUnit.SECONDS).map {
                ControlState.PLAY
            }
            ControlState.PAUSE -> Observable.just(ControlState.PAUSE)
            else -> Observable.just(ControlState.STOP)
        }
    }
}

其中fullTime 和controlCommands 是Observables，它们分别发出关于当前开始时间的事件以倒计时并说明下一步该做什么。我想链接controlCommands 并能够保持从startTime 开始的状态，当PLAY 事件出现时将倒计时，当PAUSE 出现时暂停，当STOP 出现时在startTime 重置.

scan 几乎可以工作，但我不知道如何在计时器达到 0 后停止，并且仍然每秒发送 PLAY，因为它会发送重复信息。它也不允许状态和观察值之间的分离。所以我用scan 累积的值必须与Observable 中的值类型相同（如果有意义的话）。

有什么想法我应该怎么做？

【问题讨论】：

标签： java rx-java kotlin rx-kotlin

【解决方案1】：

如果我对您的理解正确，那么您与scan 走在正确的轨道上。你只需要

.takeUntil(state -> state.hasCompleted) 在您需要停止时停止（或者可能是 takeWhile，这取决于）。
然后是.map(state -> state.getValute()) 来获取值

这有意义吗？

【讨论】：

这会导致订阅完成，还是会在新命令/时间到来时继续？
.takeUntil 将导致订阅终止 - 满足条件时它将从父级取消订阅。
所以这并不是我真正想要的，因为我想继续观察fullTime 和controlCommands。
嗯-好的，你能制作一个显示你的用例的大理石图吗？我不认为我理解您的要求。
我希望 fullTime 和 controlCommands 总是发出一些东西，因为它们来自 UI。我真正要问的是如何使用来自下游的知识（scan）暂停interval。当扫描结果为零时，我想停止在interval 上打勾，只使用零。我现在的尝试是只使用distinctUntilChanged，我认为它适用于这种情况。