c# - Rx 分组节流

Question

我有一个 IObservable<T>哪里 T 看起来像

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

通知以不同的时间间隔和不同的通知生成，其中版本号随着每个通知 ID 的每次更新而递增。

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

到目前为止，我想出了这个用于节流和分组的方法，但无法弄清楚如何实际返回 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);
}

编辑:示例输入/输出( throttle 延迟: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 }
...
...

Answer 1

该方法完全符合您的期望:

public static IObservable<Notification> ThrottledById(this IObservable<Notification> observable)
{
    return observable.Buffer(TimeSpan.FromSeconds(3))
        .SelectMany(x =>
            x.GroupBy(y => y.Id)
            .Select(y => y.Last()));
}

如果您想在第一个通知出现后收集具有相同 ID 的所有通知 n 秒并公开最后一个通知，那么您需要基于 GroupByUntil 的方法。

public static IObservable<Notification> ThrottledById(this IObservable<Notification> observable)
{
    return observable.GroupByUntil(x => x.Id, x => Observable.Timer(TimeSpan.FromSeconds(3)))
        .SelectMany(x => x.LastAsync());
}

您的样本输入/输出看起来是这样的:

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