How do you buffer items into groups in Reactive Extensions?

Question

I have an IObservable; where a property change has an entity ID and PropertyName. I want to use this to update a database, but if multiple properties change almost simultaneously I only want to do one update for all properties of the same entity.

If this was a static IEnumerable and I was using LINQ I could simply use:

MyList.GroupBy(C=>C.EntityID);

However, the list never terminates (never calls IObserver.OnComplete). What I want to be able to do is wait a period of time, say 1 second, group all the calls appropriately for that one second.

Ideally I would have separate counters for each EntityID and they would reset whenever a new property change was found for that EntityID.

I can't use something like Throttle because I want to handle all of the property changes, I just want to handle them together in one go.

Answer 1

Here you go:

MyObservable
    .Buffer(TimeSpan.FromSeconds(1.0))
    .Select(MyList =>
        MyList.GroupBy(C=>C.EntityID));