接收分组节流 [英] 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屋!