检测和测试停滞的流– RxJava常见问题解答

假设您有一个流以不可预测的频率发布事件。有时您可以预期每秒会有数十条消息，但是偶尔几秒钟都看不到任何事件。如果您的流是通过Web套接字，SSE或任何其他网络协议传输的，则可能会出现问题。静默时间过长（停顿）可以解释为网络问题。因此，我们经常不时发送人工事件（ ping ），以确保：

客户还活着
让客户知道我们还活着

举一个更具体的例子，假设我们有一个Flowable<String>流，它会产生一些事件。如果没有事件超过一秒钟，则应发送占位符"PING"消息。当静默时间更长时，应该每秒发出一个"PING"消息。我们如何在RxJava中实现这样的要求？最明显但不正确的解决方案是将原始流与ping合并：

Flowable<String> events = //...
Flowable<String> pings = Flowable.interval(1, SECONDS).map(x -> "PING");Flowable<String> eventsWithPings = events.mergeWith(pings);

mergeWith()运算符至关重要：它接受真正的events ，并将它们与恒定的ping流合并。当然，当不存在真实事件时，将显示"PING"消息。不幸的是，它们与原始流完全无关。这意味着即使有很多正常事件，我们也会继续发送ping命令。而且，当静默开始时，我们不会在一秒钟后精确发送"PING" 。如果您对这种机制感到满意，则可以在此处停止阅读。

一种更复杂的方法需要发现持续超过1秒的静音。我们可以使用timeout()运算符。不幸的是，它会产生TimeoutException并从上游退订-行为过于激进。我们只想收到某种通知。事实证明，可以使用debounce()运算符。通常，此操作员会推迟新事件的发出，以防万一有新事件出现，从而覆盖了旧事件。所以，如果我说：

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);

这意味着delayed流仅在1秒内未跟随其他事件时才会发出事件。如果events流保持足够快的速度产生事件，那么技术上delayed可能永远不会发出任何东西。我们将使用delayed流通过以下方式发现沉默：

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.map(ev -> "PING");
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

请记住， mergeWith()和它的static merge()对应物之间没有区别。所以我们到了某个地方。如果流繁忙，则delayed流将永远不会收到任何事件，因此不会发送"PING"消息。但是，当原始流不发送任何事件超过1秒时， delayed接收到最后一次看到的事件，将其忽略并转换为"PING" 。聪明，但坏了。此实现仅在发现停顿后才发送一个"PING" ，而不是每秒发送一次定期ping。很容易修复！除了将最后一次看到的事件转换为单个"PING"我们还可以将其转换为周期性ping序列：

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.flatMap(x -> Flowable.interval(0, 1, SECONDS).map(e -> "PING"));
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

您能看到缺陷在哪里吗？每当原始流中出现一点沉默时，我们就会每秒发出一次ping 。但是，一旦出现真正的事件，我们应该停止这样做。我们没有。上游的每个停顿都会导致新的无限ping流出现在最终的合并流中。我们必须以某种方式告诉pings流，因为原始流发出了真正的事件，所以它应该停止发出ping 。猜猜是什么，有takeUntil()运算符可以做到这一点！

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.flatMap(x -> Flowable.interval(0, 1, SECONDS).map(e -> "PING").takeUntil(events));
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

花一点时间完全掌握上面的代码片段。每当原始流上超过1秒没有任何反应时， delayed流就会发出一个事件。 pings流发射的序列"PING"每秒从发射每个事件的事件delayed 。但是，一旦事件出现在events流上，便会终止pings流。您甚至可以将所有这些定义为单个表达式：

Flowable<String> events = //...
Flowable<String> eventsWithPings = events.mergeWith(events.debounce(1, SECONDS).flatMap(x1 -> Flowable.interval(0, 1, SECONDS).map(e -> "PING").takeUntil(events)));

可测性

好的，我们已经编写了所有这些内容，但是我们应该如何测试事件驱动代码的这个三层嵌套的Blob？我们如何确保ping在正确的时间出现并在静音结束后停止？如何模拟各种与时间相关的场景？ RxJava具有许多杀手级功能，但是测试时间流逝可能是最大的功能。首先，让我们的ping代码更具可测试性和通用性：

<T> Flowable<T> withPings(Flowable<T> events, Scheduler clock, T ping) {return events.mergeWith(events.debounce(1, SECONDS, clock).flatMap(x1 -> Flowable.interval(0, 1, SECONDS, clock).map(e -> ping).takeUntil(events)));}

此实用程序方法采用任意的T流并添加ping ，以防该流在较长时间内不产生任何事件。我们在测试中像这样使用它：

PublishProcessor<String> events = PublishProcessor.create();
TestScheduler clock = new TestScheduler();
Flowable<String> eventsWithPings = withPings(events, clock, "PING");

哦，男孩， PublishProcessor ， TestScheduler ？ PublishProcessor是一个有趣的类，它是一个亚型Flowable （所以我们可以使用它作为一个普通的流）。另一方面，我们可以使用其onNext()方法强制发出事件：

events.onNext("A");

如果有人收听events流，他将立即收到"A"事件。这clock是怎么回事？ RxJava中以任何方式处理时间的每个运算符（例如debounce debounce() ， interval() ， timeout() ， window() ）都可以采用可选的Scheduler参数。它充当时间的外部来源。特殊的TestScheduler是我们完全控制的人为时间来源。也就是说，只要我们不显式调用advanceTimeBy()时间就保持静止：

clock.advanceTimeBy(999, MILLISECONDS);

999毫秒不是巧合。 Ping在1秒钟后开始精确显示，因此在999毫秒后将不可见。现在是时候揭示完整的测试用例了：

@Test
public void shouldAddPings() throws Exception {PublishProcessor<String> events = PublishProcessor.create();final TestScheduler clock = new TestScheduler();final Flowable<String> eventsWithPings = withPings(events, clock, "PING");final TestSubscriber<String> test = eventsWithPings.test();events.onNext("A");test.assertValues("A");clock.advanceTimeBy(999, MILLISECONDS);events.onNext("B");test.assertValues("A", "B");clock.advanceTimeBy(999, MILLISECONDS);test.assertValues("A", "B");clock.advanceTimeBy(1, MILLISECONDS);test.assertValues("A", "B", "PING");clock.advanceTimeBy(999, MILLISECONDS);test.assertValues("A", "B", "PING");events.onNext("C");test.assertValues("A", "B", "PING", "C");clock.advanceTimeBy(1000, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING");clock.advanceTimeBy(999, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING");clock.advanceTimeBy(1, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING", "PING");clock.advanceTimeBy(999, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING", "PING");events.onNext("D");test.assertValues("A", "B", "PING", "C", "PING", "PING", "D");clock.advanceTimeBy(999, MILLISECONDS);events.onNext("E");test.assertValues("A", "B", "PING", "C", "PING", "PING", "D", "E");clock.advanceTimeBy(999, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING", "PING", "D", "E");clock.advanceTimeBy(1, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING", "PING", "D", "E", "PING");clock.advanceTimeBy(3_000, MILLISECONDS);test.assertValues("A", "B", "PING", "C", "PING", "PING", "D", "E", "PING", "PING", "PING", "PING");
}

看起来像一堵墙，但这实际上是我们逻辑的完整测试方案。它可以确保ping在1000毫秒后精确显示，在寂静时间很长的情况下会重复执行，而在出现真正的事件时会重复执行。但最重要的部分是：该测试是100％可预测的并且非常快。没有Awaitility ，忙等待，轮询，间歇性测试失败和缓慢。我们完全控制的人工时钟可确保所有这些组合流均按预期工作。