Rx Extensions сбрасывают оставшиеся предметы в Buffered наблюдаемые

Question

Итак, у меня есть этот код:

ISubject<int> _processed = new ReplaySubject<int>();
_processed.Buffer(5000).Subscribe(UpdateProcessed);

// Start some process which calls _processed.OnNext

Проблема, с которой я сталкиваюсь, заключается в том, что иногда буфер не заполняется, потому что последний пакет меньше 5000 и процесс завершается без вызова UpdateProcessed.Выполнено.

Есть ли способ сбросить оставшиеся элементы в _processed, наблюдаемом после завершения обработки?

Answer 1

OnComplete очистит буферы.

Попробуйте _processed.OnComplete();

Если вы не знаете, когда процесс завершится, вы всегда можете добавить временной интервал в буфер.Он будет генерировать элементы, когда истекает промежуток времени или когда буфер заполнен, в зависимости от того, что наступит раньше.

_processed.Buffer(TimeSpan.FromSeconds(1), 5000).Subscribe(UpdateProcessed);