如何在响应式扩展中将项缓冲到组中

Question

我有一个IObservable;其中属性更改具有实体ID和PropertyName。我想用它来更新数据库，但是如果多个属性几乎同时改变，我只想对同一实体的所有属性做一次更新。

如果这是一个静态IEnumerable，并且我正在使用LINQ，我可以简单地使用:

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

然而，列表永远不会终止(永远不会调用IObserver.OnComplete)。我想要做的是等待一段时间，比如1秒，在这一秒内对所有呼叫进行适当的分组。

理想情况下，我将为每个EntityID设置单独的计数器，当发现该EntityID的新属性更改时，它们将重置。

我不能使用像Throttle这样的东西，因为我想处理所有的属性更改，我只想一次性处理它们

Answer 1

给你:

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

Answer 2

Buffer方法似乎可以满足您的要求。给它一个TimeSpan，它会把所有的消息折叠成一个列表。还有Window方法，它做了类似的事情，但我不完全确定它的语义可能是什么。