System.Reactive: Buffer(timeSpan, timeShift) by Timestamp field

问题描述

如 this previous question 中所述，我在 C# 项目中使用 Reactive 库根据预先配置的策略对传入数据进行分组。所有这些数据都实现了以下接口：

public interface IPoint
{
    object Value { get; }
    DateTimeOffset Timestamp { get; }
}

我的目标是根据接收到的数据的 Timestamp 实现“跳跃”缓冲区（缓冲区大小和跳跃/移位大小都在开头声明为 TimeSpan）。跃点/移位大小可以小于缓冲区大小，这意味着某些 IPoint 实例可以属于多个组。

一个例子：考虑以下IPoint

Value: 1,Timestamp: "2021-05-25T00:00:01"
Value: 2,Timestamp: "2021-05-25T00:00:02"
Value: 3,Timestamp: "2021-05-25T00:00:03"
Value: 4,Timestamp: "2021-05-25T00:00:04"
Value: 5,Timestamp: "2021-05-25T00:00:05"

• 缓冲区大小为 3 秒，跳跃/移位大小为 2 秒，我希望它们被分组为 [1,2,3],[3,4,5]。
• 缓冲区大小为 2 秒，跳跃/移位大小为 3 秒，我希望它们被分组为 [1,2],[4,5]

我已经看到有一个 Buffer(timeSpan,timeShift) 扩展来完成这项工作，但它考虑的是运行时计算的时间戳，而不是传递的 IPoint 的时间戳。

我试图寻找解决方案，但找不到任何有用的东西。

我是 Reactive 的新手，所以欢迎提出任何有用的评论（也适用于其他问题）。谢谢。

编辑：与上一个问题一样，我以这种方式使用 ISubject<IPoint>：

ISubject<IPoint> subject = new Subject<IPoint>();
// ...
// when new data come from an external source
public void Add(IPoint newPoint)
{
    subject.OnNext(newPoint);
}

// subscription made by another class in order to be called when "hopping" buffer is ready
public void Subscribe(Action<IEnumerable<IPoint>> callback)
{
    // Todo: implement buffer + .Subscribe(callback)
}

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）