接收分组节流 [英] Rx grouped throttling

问题描述

我有一个 IObservable< T> ，其中T看起来像

I have an IObservable<T> where T looks like

public class Notification
{
    public int Id { get; set; }
    public int Version { get; set; }
}

通知会在不同的时间间隔生成并针对不同的通知，其中版本号会随着每个通知ID的每次更新而递增.

Notifications are produced at variable time intervals and for different notifications, where the version number get incremented with each updated per notification id.

什么是在特定时间段内限制可观察对象然后接收带有最新版本"字段的明确通知的正确方法?

What would be a proper approach to throttle the observable for a specific period of time and then to receive distinct notifications with the latest Version field?

到目前为止，我想出了此方法来进行限制和分组，但无法弄清楚如何实际返回 IObservable< Notification> .

So far I came up with this for throttling and grouping, but can't figure out how to actually return IObservable<Notification>.

public static IObservable<int> ThrottledById(this IObservable<Notification> observable)
{
    return observable
        .GroupByUntil(n => n.Id, x => Observable.Timer(TimeSpan.FromSeconds(1)))
        .Select(group => group.Key);
}

样本输入/输出(油门延迟:3):

Sample input/output (Throttle delay: 3):

1. { id: 1, v: 1 }
2. { id: 1, v: 2 }  { id: 2, v: 1 }
3. { id: 1, v: 3 }
-----------------------------------> notify { id:1, v: 3 }, notify { id:2, v: 1 }
4. 
5. { id: 2, v: 2 }
6.
-----------------------------------> notify { id:2, v: 2 }
7. { id: 1, v: 4 }
8. { id: 1, v: 5 }  { id: 2, v: 3 }
9. { id: 1, v: 6 }
-----------------------------------> notify { id:1, v: 6 }, notify { id: 2, v: 3 }
...
...

推荐答案

这似乎会产生所需的输出

This seems to produce your desired output

 IObservable<Notification> GroupByIdThrottle(IObservable<Notification> producer, IScheduler scheduler)
    {
        return Observable.Create<Notification>(observer =>
        {
            return producer
                .GroupByUntil(
                    item => item.Id,
                    item => Observable.Timer(TimeSpan.FromSeconds(3), scheduler))
                .SelectMany(result =>
                {
                    return result.Aggregate<Notification, Notification>(null, (dict, item) =>
                    {
                        return item;
                    });
                })
                .Subscribe(observer);

        });
    }

这样的想法是，聚合使它成为可能，因此只有每组的最后一个值才能使它活着，并且一旦计时器达到3秒，分组的流就完成了.

The idea being that the aggregate makes it so only the last value of each group makes it out alive and the grouped streams complete once the timer hits 3 seconds.

我食人了

想法

这篇关于接收分组节流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！