java - 如何访问 rx 中先前发射的数据？

Question

问题是关于rx 编程的。我有一个代码:

Observable.from(array)
          .map(array_item1 -> array_item2)
          .map(array_item3 -> array_item4)
          .map(array_item5 -> array_item6)

在第三个 map 调用中，我不仅需要访问 array_item5，还需要访问 array_item1 或 array_item3。它应该是这样的:

...    
.map(array_item5 -> array_item5 + array_item1)
...

是否有任何特殊的模式或运算符？

Answer 1

如果您想访问以前的排放量，您需要将它们保存在某个地方。有累加器运算符:

收集 - 收集单个可变结构中的所有排放量

Observable.from(Arrays.asList(1,2,3,4,5,6))
      .collect(ArrayList::new,
        (array, item) -> {
            //access previous and current emissions
            array.add(item);
        })

扫描 - 除了收集之外，它还会在接下来的每个上返回所有先前的排放:

Observable.from(Arrays.asList(1,2,3,4,5,6))
    .scan(new ArrayList<>(), (array, item) -> {
           array.add(item)     
           return array;
          })
    .map(collectedItems -> {
          //access all previous and curent emissions on every new
          return collectedItems;
        })