与 Rx 运行总和

Question

一定有人已经解决了这个问题。 假设我有一个类定期引发有关值更改的事件（例如 PropertyChanged） 这个价值无非就是金钱的数量。

现在，我想利用 Rx 来获得最后 10 分钟的增量总和。例如 BufferWithTime 没有帮助，因为我总是需要最后 10 分钟。

我有什么想法可以做到这一点吗？

蒂亚 马丁

Answer 1

下面的解决方案涉及使用 Observable.Scan 将前十分钟内相关事件数据的状态保留在列表中。状态以元组列表的形式维护，并以 int（金钱）和 DateTime 作为值。

var events = Observable.FromEvent<YourEventArgs>(
    h => SomeEvent += h, h => SomeEvent -= h);
var runningSums =
    events.Scan(new List<Tuple<int, DateTime>>(),
                (l, e) =>
                {
                    var now = DateTime.Now;
                    // Add last event data to list.
                    l.Add(Tuple.Create(e.EventArgs.Money, now));
                    // Return the correct part of the list (everything
                    // from the last ten minutes).
                    return l.Where(t => (now - t.Item2) <
                                   TimeSpan.FromMinutes(10)).ToList();
                 })
          .Select(l => l.Sum(t => t.Item1));
runningSums.Subscribe(sum => Console.WriteLine(sum));

编辑：不为每个事件返回新列表的示例：

var events = Observable.FromEvent<YourEventArgs>(
    h => SomeEvent += h, h => SomeEvent -= h);
var runningSums =
    events.Scan(Tuple.Create(new List<Tuple<int, DateTime>>(),
                             DateTime.Now - TimeSpan.FromMinutes(10)),
                (l, e) =>
                {
                    var now = DateTime.Now;
                    l.Item1.Add(Tuple.Create(e.EventArgs.Nr, now));
                    // if (trimming-condition) then trim front of list...
                    return Tuple.Create(l.Item1, now - TimeSpan.FromMinutes(10));
                })
          .Select(l => l.Item1.Where(t => t.Item2 > l.Item2).Sum(t => t.Item1));
runningSums.Subscribe(sum => Console.WriteLine(sum));

Answer 2

好吧，看看下面的解决方案。它建立在之前提出的解决方案的基础上，但为了效率（和可读性，我认为）而放弃了纯函数风格。它还重用内置类型 Timestamped 来跟踪时间...

干杯

    public static class RxEntentsions
        {
            class TimeLimitedList<T>
            {
                public List<Timestamped<T>> Values = new List<Timestamped<T>>();
                TimeSpan span;
                public TimeLimitedList(TimeSpan sp) { span = sp; }
                public void Add(Timestamped<T> v)
                {
                    Values.Add(v);
                    Values.RemoveAll(a => a.Timestamp < (DateTime.Now - span));
                }
            }

            public static IObservable<List<Timestamped<TSource>>> SlidingWindow<TSource>(this IObservable<Timestamped<TSource>> source, TimeSpan slidingWindow)
            {
                return source.Scan0(new TimeLimitedList<TSource>(slidingWindow), (acc, v) => { acc.Add(v); return acc; }).Select(a => a.Values);
            }
        }


    static void Main(string[] args)
    {
        var gen = Observable.Interval(TimeSpan.FromSeconds(0.25d)).Timestamp();
        gen.SlidingWindow(TimeSpan.FromSeconds(1)).Subscribe(slw => {slw.ForEach(e=> Console.WriteLine(e)); Console.WriteLine("--------");});
        Console.ReadLine();
    }