Rx 扩展刷新 Buffered observable 中的剩余项目

Question

所以我有这个代码：

ISubject<int> _processed = new ReplaySubject<int>();
_processed.Buffer(5000).Subscribe(UpdateProcessed);

// Start some process which calls _processed.OnNext

我遇到的问题是有时缓冲区没有填满，因为最后一批小于 5000 并且进程在没有调用UpdateProcessed执行的情况下退出。

_processed处理完成后有没有办法刷新observable中剩余的项目？

Answer 1

OnComplete 将刷新缓冲区。

尝试 _processed.OnComplete();

如果您不知道该过程何时完成，您可以随时向缓冲区添加一个时间跨度。当时间跨度到期或缓冲区已满时，它会发出项目，以先到者为准。

_processed.Buffer(TimeSpan.FromSeconds(1), 5000).Subscribe(UpdateProcessed);