反应式扩展：事件的频率

Question

假设我有一个像 MyClass { string id, string eventType, datetime ts}

ts 是事件的时间戳，Id 是我要计算频率的时间

我有一个热门的 MyClass 可观察对象，我想计算过去 30 秒内每个 stringId 的 recvd 事件数

如果事件数超过 5，我会引发另一个 MyClass 事件（具有相同的 Id，并且 eventType =“New”），如果它再次低于 3，我需要更新之前引发的事件（具有相同的Id 和 eventType ="New" )。

我觉得我需要使用滑动窗口，我到现在为止

public static IObservable<MyClass> CountFrequency(this IObservable<MyClass> source, TimeSpan withinPeriod, string marker)
{
    // var scheduler = new HistoricalScheduler();
    //  var driveSchedule = source.Subscribe(e => scheduler.AdvanceTo(e.Timestamp)); 
    return source.Window(TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(5))
        .SelectMany(sl => sl)
        .GroupBy(a => a.id)
        .SelectMany(go => go
        .Aggregate(new MyClass(), (acc, evt) => CustomAggFrequency(acc, evt, marker))
        .Select(count => count)));
}

看不懂

a) 如何将调度程序与数据的时间戳而非系统时间相关联 b) 如何编写CustomAggFrequency的逻辑

任何建议

Answer 1

这行不通吗？您的问题的措辞很难弄清楚您想要什么。

var span = TimeSpan.FromSeconds(30);
var shift = TimeSpan.FromSeconds(5);
var query = source
    .Window(span, shift)
    .Select(window => window
        .GroupBy(item => item.id)
        .ToDictionary(g => x.Key, g => g.Count() / span.TotalSeconds));

此query 将每 5 秒发出一个字典，将每个 id 映射到其在过去 30 秒内的频率（以赫兹为单位）。它的类型是IObservable<Dictionary<string, double>>。