RxJava缓冲-忽略零项

Question

下面是我用来缓冲和转换传入事件的代码：

public Publisher<Collection<EventTO>> logs(String eventId) {
    ConnectableObservable<Event> connectableObservable = eventsObservable
        .share().publish();
    connectableObservable.connect();

    connectableObservable.toFlowable(BackpressureStrategy.BUFFER)
        .filter(event -> event.getId().equals(eventId))
        .buffer(1, TimeUnit.SECONDS, 50)
        .map(eventsMapper::mapCollection);
}

这里的问题是，Flowable每秒返回一个空列表，尽管没有发布到eventsObservable的事件。

是否有一种方法来保持。缓冲区直到至少有一个对象？

注意：看起来有办法在C#中做到这一点（这里描述：https://stackoverflow.com/a/30090185/668148）。但是Java怎么做呢？

Answer 1

正如马克·基恩所建议的那样，。distinctUntilChanged起作用。

因此，如果缓冲后有1项，以下代码将推送事件列表：

connectableObservable.toFlowable(BackpressureStrategy.BUFFER)
    .filter(event -> event.getId().equals(eventId))
    .buffer(1, TimeUnit.SECONDS, 50)
    .distinctUntilChanged()             // <<<======  
    .map(eventsMapper::mapCollection);