相关疑难解决方法(0)

鉴于IObservable<T>是否有一种使用Throttle行为的方法(在添加项目时重置计时器,但让它返回在该时间内添加的所有项目的集合？

Buffer它提供了类似的功能,它可以IList<T>在每个时间跨度或计数时将数据分块.但是每次添加项目时我都需要重置时间.

我在这里看到过类似的问题,反应式扩展是否支持滚动缓冲区？,但答案看起来并不理想,而且有点旧,所以我想知道Rx-Main的发布版现在是否支持这个功能.

我的问题有点像Nagle算法创建的问题,但不完全正确.我想要的是将OnNext通知缓冲IObservable<T>到一系列IObservable<IList<T>>s中,如下所示:

1. 第一个T通知到达时,将其添加到缓冲区并开始倒计时
2. 如果T在倒计时到期之前收到另一个通知,请将其添加到缓冲区并重新开始倒计时
3. 一旦倒计时结束(即生产者已经沉默了一段时间),将所有缓冲的T通知转发为单个聚合IList<T>通知.
4. 如果缓冲区大小在倒计时到期之前超过某个最大值,则无论如何都要发送它.

IObservable<IList<T>> Buffer(this IObservable<T>, Timespan, int, IScheduler) 看起来很有希望,但它似乎定期发送聚合通知,而不是"在第一个通知到达时启动计时器,当其他通知到达时重启它"我想要的行为,并且它还发送一个空列表在如果没有从下面生成通知,则每个时间窗口的结束.

我不希望放弃任何的T通知; 只是缓冲它们.

有这样的事情存在,还是我需要自己编写？