Reactive: Buffer with aggregate function

[Thaina]

Feature request

Currently now buffer can only batch a bunch of object as an array to pass to next operator. But sometimes we don't actually need all those values. Instead we might have custom logic to pick value or summarize value that arrived into buffer

So I think we should have operator like buffer that also take a parameter in reduce manner

Suppose I would want the minimum by some value of the object from buffer. Normally I could do this

observable.Buffer(() => onClickEventStream).Select((buffer) => {
    var seed = buffer.FirstOrDefault();
    foreach(var obj in buffer.Skip(1))
    {
        if(obj?.value < seed?.value)
            value = seed;
    }
    return value;
});

But this method would create unnecessary array and cache many values that not being used

So I would like to propose

IObservable<TBufferClosing> Buffer<TSource, TBufferClosing>(this IObservable<TSource> source, Func<IObservable<TBufferClosing>> bufferClosingSelector,Func<TBufferClosing, TSource, TBufferClosing> aggregator);

//// usage

observable.Buffer((_) => onClickEventStream.Select((_) => null),(seed,obj) => {
    return obj?.value < seed?.value ? obj : seed;
});

This way we could have any custom accumulation logic without the need to create array. We might output dictionary or do math on some value freely

[quinmars]

You could combine the Window operator with an aggregating operator, e.g., Min, Max, Sum, Aggregate, etc. I haven't tested it but the following should meet your expectations:

observable.Window(() => onClickEventStream).SelectMany(window => window.Min())

[Thaina]

@quinmars I think both window and buffer contain the same problem

It would make an array, or a list, to collect things in each buffer (or each window) into batch before continue isn't it?

What's what I don't want. I think we could aggregate everything, use reduce function, when the buffer are still buffering

Suppose the window are too long, user have left away from keyboard and leave the window opened for hour. There might be 1000 object waiting in the queue for the interaction from user, it then take memory and make large memory footprint. While if we can aggregate it there will be only one item all the times

[quinmars]

The Window operator does not buffer. It pushes the elements forward on arrival. Observable.Min does not buffer items either.

[Thaina]

@quinmars Could I assume that buffer is actually window with the logic of list.Add ?

[bartdesmet]

Yes, the following identity holds more or less:

xs.Buffer(a, b) == xs.Window(a, b).SelectMany(w => w.ToListAsync())

The answer here is to use composition using Window and aggregate functions, as @quinmars suggested. As such, I don't see a need to add more specialized operators to Rx in this case.