c# - react 性扩展 : buffer until subscriber is idle

Question

我有一个程序，我在其中接收事件并希望分批处理它们，以便在我处理当前批处理时进入的所有项目都将出现在下一批处理中。

Rx 中简单的 TimeSpan 和基于计数的 Buffer 方法会给我多批项目，而不是给我一大批已经进来的所有东西(如果订阅者花费的时间超过指定的 TimeSpan 或超过 N 个项目进来，N 大于计数)。

我研究了使用更复杂的 Buffer 重载 Func<IObservable<TBufferClosing>>或 IObservable<TBufferOpening> and Func<TBufferOpening, IObservable<TBufferClosing>> ，但我找不到如何使用它们的示例，更不用说弄清楚如何将它们应用于我正在尝试做的事情了。

Answer 1

这是否符合您的要求？

var xs = new Subject<int>();
var ys = new Subject<Unit>();

var zss =
    xs.Buffer(ys);

zss
    .ObserveOn(Scheduler.Default)
    .Subscribe(zs =>
    {
        Thread.Sleep(1000);
        Console.WriteLine(String.Join("-", zs));
        ys.OnNext(Unit.Default);
    });

ys.OnNext(Unit.Default);
xs.OnNext(1);
Thread.Sleep(200);
xs.OnNext(2);
Thread.Sleep(600);
xs.OnNext(3);
Thread.Sleep(400);
xs.OnNext(4);
Thread.Sleep(300);
xs.OnNext(5);
Thread.Sleep(900);
xs.OnNext(6);
Thread.Sleep(100);
xs.OnNext(7);
Thread.Sleep(1000);

我的结果:

1-2-3
4-5
6-7