Create new buffer every minute

Question

Given an observable sequence with values containing a timestamp how do I buffer the sequence into new buffers each containing one minute of data? A buffer must close exactly when a new minute is started..

Thanks

Answer 1

Assuming your sequence of values is already sorted by timestamp, you can use GroupByUntil to group them by their minute. In this example, I then turn each group into a List:

// Utility method from:
// http://stackoverflow.com/a/10100259/674326
DateTime RoundDown(DateTime dt, TimeSpan d)
{
    return new DateTime((dt.Ticks / d.Ticks) * d.Ticks);
}

//...
var sequence = // ...
var interval = TimeSpan.FromMinutes(1);
var intervalBuffers = sequence
    .Publish(items => items.GroupByUntil(
        item => RoundDown(item.TimeStamp, interval),
        grp => items.Where(item => grp.Key != RoundDown(item.TimeStamp, interval))))
    .Select(grp => grp.ToList());