.net - 如何在 RX 中实现超时缓冲

Question

我需要实现一个事件处理，当一段时间内没有新事件到达时延迟完成。 (当文本缓冲区更改时，我必须排队解析任务，但我不想在用户仍在键入时开始解析。)

我是 RX 的新手，但据我所知，我需要 BufferWithTime 和 Timeout 方法的组合。我想这是这样工作的:它缓冲事件，直到在后续事件之间的指定时间段内定期收到它们。如果事件流中存在间隙(长于时间跨度)，它应该返回传播到目前为止缓冲的事件。

看看Buffer和Timeout是怎么实现的，大概可以实现我的BufferWithTimeout方法(如果大家有的话，请分享给我)，不知道是不是结合现有的方法就可以实现。有任何想法吗？

Answer 1

这是一个相当古老的问题，但我相信以下答案值得一提，因为所有其他解决方案都迫使用户手动订阅、跟踪更改等。

我提供以下作为“Rx-y”解决方案。

var buffers = source
    .GroupByUntil(
        // yes. yes. all items belong to the same group.
        x => true,
        g => Observable.Amb<int>(
               // close the group after 5 seconds of inactivity
               g.Throttle(TimeSpan.FromSeconds(5)),
               // close the group after 10 items
               g.Skip(9)
             ))
    // Turn those groups into buffers
    .SelectMany(x => x.ToArray());

基本上，源是窗口化的，直到根据最新窗口定义了一些可观察的。创建了一个新窗口(分组的 observable)，我们使用该窗口来确定该窗口应该何时关闭。在这种情况下，我在 5 秒不活动或最大长度为 10 (9+1) 后关闭窗口。