Reactive Extensions에서 항목을 그룹으로 버퍼링하는 방법은 무엇입니까?

Question

나는 IObservable을 가지고있다; 여기서 속성 변경에는 엔터티 ID와 PropertyName이 있습니다. 이 방법을 사용하여 데이터베이스를 업데이트하려고하지만 여러 속성이 거의 동시에 변경되는 경우 동일한 엔터티의 모든 속성에 대해 하나의 업데이트 만 수행하려고합니다. 이 정적는 IEnumerable이었고, 나는 간단하게 사용할 수있는 LINQ를 사용하는 경우

:

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

그러나 목록이 종료되지 않습니다 (IObserver.OnComplete 호출되지 않습니다). 내가 할 수 있기를 원하는 것은 1 초의 시간을 기다려서 1 초 동안 모든 호출을 적절하게 그룹화하는 것입니다.

이상적으로 나는 각 EntityID에 대해 별도의 카운터를 가지며 해당 EntityID에 대한 새 속성 변경이 발견 될 때마다 다시 설정됩니다.

나는 모든 속성 변경을 처리하기 위해 Throttle과 같은 것을 사용할 수 없으며 한 번에 함께 처리하려고합니다.

Answer 1

당신은 이동 :

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

Answer 2

Buffer 메서드는 원하는대로하는 것처럼 보입니다. TimeSpan을 지정하면 모든 메시지가 목록으로 축소됩니다. 비슷한 것을하는 Window 메서드가 있지만, 그 의미가 무엇인지 모를 정도입니다. 여기