Buffer operator with count and time conditions

Question

I have a sequence that is extremely chatty, and I was trying to make it more efficient by processing events in batches. A Buffer operator with time and count conditions was something that seemed to match my requirements, except for one small nuance. When you use this overload, the subscription gets notified after the specified time delay, regardless whether there are any items in the buffer. This gets really annoying cause most of the time my subscription gets an empty list from the buffer operator. Considering that it is a multi-threaded application where subscriber is on UI thread, it turns out to be not the most optimal approach to process items in batches. I was wondering if there was a way to use available operators to create a sequence that would fire either when a certain amount of items in the buffer are present, or when a certain time has passed, but if and only if there are any items in the buffer. I know that I could do something like this:

sequence.Buffer(TimeSpan.FromSeconds(5), 1).Where(e=>e.Count > 0)

But I was wondering if there is another way to do this, cause somehow I feel that it's not the best way.

Answer 1

I can't see a reason to worry about this - you have an idiomatic solution. An empty buffer is information, so it's reasonable for the framework implementation to return it. Any other method would effectively be doing the same thing you are internally anyway.

When I find myself using small groups of standard operators I often wrap them in a more explanatory extension method. E.g.:

public static class ObservableExtensions
{
    public static IObservable<IList<T>> ToNonEmptyBuffers<T>(
        this IObservable<T> source,
        TimeSpan timespan,
        int count,
        IScheduler scheduler = null)
    {
        scheduler = scheduler ?? Scheduler.Default;
        return source.Buffer(timespan, count, scheduler ?? Scheduler.Default)
                     .Where(buffer => buffer.Count > 0);
    }
}

Allowing:

sequence.ToNonEmptyBuffers(TimeSpan.FromSeconds(5), 1);