Rx Rolling Buffer

gistfile1.txt

public static class AvaloniaObservableExtensions
    {
        public static IObservable<IList<T>> RollingBuffer<T>(this IObservable<Timestamped<T>> @this, TimeSpan buffering)
        {
            return Observable.Create<IList<T>>(o =>
            {
                var list = new LinkedList<Timestamped<T>>();

                return @this.Subscribe(tx =>
                {
                    list.AddLast(tx);

                    while (list.First.Value.Timestamp < DateTime.Now.Subtract(buffering))
                    {
                        list.RemoveFirst();
                    }

                    o.OnNext(list.Select(tx2 => tx2.Value).ToList());
                }, ex => o.OnError(ex), () => o.OnCompleted());
            });
        }