In the ongoing saga of me rewriting functions that can be easily composed from existing Rx functions, I present my latest progeny: a bastard combination of Scan and Aggregate. The aim is to aggregate n items from a source, and emit the result every n. Then start again.public static IObservable<T> BufferAggregate<T>(this IObservable<T> source,                                                 Func<T, T, T> aggregator, int size){    return Observable.Create<T>(observer =>    {        var count = 0;        var current = default(T);         return source.Subscribe(t =>        {            count++;            if (count == 1)                current = t;            else if (count < size)                current = aggregator(current, t);            else // (count == size)            {                observer.OnNext(aggregator(current, t));                count = 0;            }        });    });}Note that there is a good reason for not using pure Buffer, which is that the data items are rather large, so it is preferable to calculate the aggregate as we go along, rather than doing it all at the end.Am I complicating things too much?
BufferAggregate
c#;system.reactive