RxJava2，如何在每次发射时发出迄今为止所有发射的列表答案

【问题标题】：RxJava2, how to emit a list of all emits so far on each emitRxJava2，如何在每次发射时发出迄今为止所有发射的列表
【发布时间】：2019-04-26 23:54:22
【问题描述】：

我有一个 Observable 发出的对象流。

每次有发射时，我都想将其映射到目前为止所有发射的列表，

我知道我可以通过存储全局列表并手动执行此操作，但我想知道是否有开箱即用的操作员。

a -> [a]
b -> [a,b]
c -> [a,b,c]
a -> [a,b,c,a]
d -> [a,b,c,a,d]
f -> [a,b,c,a,d,f]
b -> [a,b,c,a,d,f,b]
g -> [a,b,c,a,d,f,b,g]

当我使用 toList() 时。一旦流完成，它只会发送一个列表。

编辑 1：这是我的流程现在的样子：

source.itemStream()
     .doOnNext( item -> handleItem())
     .subscribe()

而我更喜欢的是：

source.itemStream()
     .someOperator() // this will map to a stream of all items so far.
     .doOnNext( item -> handleItem())
     .subscribe()

或者

source.itemStream()
     .compose(x -> listBuildingFunction(x)) // this will map to a stream of all items so far.
     .doOnNext( item -> handleItem())
     .subscribe()

【问题讨论】：

标签： rx-java2

【解决方案1】：

将项目收集到一个列表中并发出该列表的副本：

Observable.defer(() -> {
    List<Integer> seen = new LinkedList<>();
    return Observable.range(1, 10)
           .map(v -> {
                seen.add(v);
                return new ArrayList<>(seen);
           });
})
.subscribe(System.out::println);

【讨论】：

【解决方案2】：

使用上面的 akarnokd 的答案，我做了以下事情：（在 Kotlin 中）

private fun toPartialList(): ObservableTransformer<Item>, List<Item>> {
    return ObservableTransformer { observable ->
        Observable.defer { 
            val tempList = ArrayList<Item>()
            observable.map { item ->
                tempList.add(item)
                return@map ArrayList(tempList)
            }
        }
    }
}

然后在我的直播中使用它：

source.itemStream()
     .compose(toPartialList()) 
     .doOnNext { handleItems(it) }
     .subscribe()

【讨论】：