How do you buffer items into groups in Reactive Extensions?

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 s开发者_运维问答eparate 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.

---

Here you go:

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

---

The Buffer method seems to do what you want. Give it the TimeSpan and it'll collapse all the messages into a list. There is also the Window method which does something similar, but I'm not entirely sure what its semantics might be.