使用反应扩展每秒按顺序获取下一个事件

我有以下类型...

public class NewsFeed
{
    public event EventHandler<NewsItemEventArgs> NewItem;

    .....

}

public class NewsItemEventArgs : EventArgs
{
    public NewsItem Item;
    public NewsItemEventArgs(NewsItem newsItem)
    {
        Item = newsItem;
    }
}

public class NewsItem
{
    public int Id { get; set; }
    public string Title { get; set; }
    public string Body { get; set; }
}

NewsFeed 的 NewItem 事件触发带有 NewsItemEventArgs 类型的 eventArgs 的事件。在我的系统中，事件会突发发布，例如在 1 秒的小窗口中发布 10 个新闻项目，然后在 60 秒内不再发布新闻报道。我想用 RX 来平滑这些爆发，这样我的 UI 新闻故事“看起来”就会以更规则的间隔（例如 5 秒）一次到达一个。

我知道我需要创建一个可观察的对象，如下所示

_source = Observable.FromEventPattern<NewsItemEventArgs>(
                        h => _newsFeed.NewItem += h,
                        h => _newsFeed.NewItem -= h);

但我不知道如何转换和订阅可观察到的事件，这样我就可以滴灌事件，而不是像上面描述的那样突发事件。

有任何想法吗？

Zip 可能不是此操作的最佳选择，因为生产者有时可能会很慢，从而导致输出不稳定。

看来是准确的调度与DateTimeOffsetRx 2.0 仍然不可能。TimeSpan不过目前还可以。您可以通过更换来尝试TimeSpan偏移量DateTimeOffset.

综上所述，如果我们能够指定两个连续值之间的最小间隔，我们就可以解决突发问题。

    static IObservable<T> DelayBetweenValues<T>(this IObservable<T> observable, TimeSpan interval, IScheduler scheduler)
    {
        return Observable.Create<T>(observer =>
        {
            var offset = TimeSpan.Zero;
            return observable
                .TimeInterval(scheduler)
                .Subscribe
                (
                    ts =>
                    {
                        if (ts.Interval < interval)
                        {
                            offset = offset.Add(interval);
                            scheduler.Schedule(offset, () => observer.OnNext(ts.Value));
                        }
                        else
                        {
                            offset = TimeSpan.Zero;
                            observer.OnNext(ts.Value);
                        }
                    }
                );
        });
    }

Test:

        Observable.Interval(TimeSpan.FromSeconds(2.5))
                  .Do(_ => Console.WriteLine("Burst"))
                  .SelectMany(i => Enumerable.Range((int)i, 10))
                  .DelayBetweenValues(TimeSpan.FromSeconds(0.2), TaskPoolScheduler.Default)
                  .Subscribe(Console.WriteLine);