c# - 如何使用 RX 限制事件流？

Question

我想有效地限制事件流，以便在收到第一个事件时调用我的委托(delegate)，但如果收到后续事件则不会持续 1 秒。在该超时(1 秒)到期后，如果收到后续事件，我希望我的委托(delegate)被调用。

是否有一种简单的方法可以使用 Reactive Extensions 来做到这一点？

示例代码:

static void Main(string[] args)
{
    Console.WriteLine("Running...");

    var generator = Observable
        .GenerateWithTime(1, x => x <= 100, x => x, x => TimeSpan.FromMilliseconds(1), x => x + 1)
        .Timestamp();

    var builder = new StringBuilder();

    generator
        .Sample(TimeSpan.FromSeconds(1))
        .Finally(() => Console.WriteLine(builder.ToString()))
        .Subscribe(feed =>
                   builder.AppendLine(string.Format("Observed {0:000}, generated at {1}, observed at {2}",
                                                    feed.Value,
                                                    feed.Timestamp.ToString("mm:ss.fff"),
                                                    DateTime.Now.ToString("mm:ss.fff"))));

    Console.ReadKey();
}

当前输出:

Running...
Observed 064, generated at 41:43.602, observed at 41:43.602
Observed 100, generated at 41:44.165, observed at 41:44.602

但我想观察(时间戳显然会改变)

Running...
Observed 001, generated at 41:43.602, observed at 41:43.602
....
Observed 100, generated at 41:44.165, observed at 41:44.602

Answer 1

好的，

这里有 3 个场景:

1) 我想每秒获取事件流的一个值。意思是:如果它每秒产生更多的事件，您将始终获得更大的缓冲区。

observableStream.Throttle(timeSpan)

2) 我想获得最新的事件，它是在第二次发生之前产生的意味着:其他事件被丢弃。

observableStream.Sample(TimeSpan.FromSeconds(1))

3) 您想获取最后一秒发生的所有事件。并且每一秒

observableStream.BufferWithTime(timeSpan)

4) 你想选择在第二个与所有值之间发生的事情，直到第二个过去，然后返回你的结果

observableStream.CombineLatest(Observable.Interval(1000), selectorOnEachEvent)