rxjs - 使用 RxJS 操作项目流以获得流式项目数组

Question

我是 rxjs 的新手，无法解决这个问题:

我有两个流:

• 一个有传入的对象
  • ---a----b----c----d----->
• 一个是从列表中选择的对象
  • ----------------c---->

从传入的对象流中生成对象列表流(使用扫描运算符)

  incoming: ----a--------b-------c----------d----------------\>
  list:  -------[a]----[a,b]----[a,b,c]----[a,b,c,d]---------\>

当一个列表对象被选中(n)时，开始一个新的流

• 新流的第一个值是切片列表的最后一个值(list.slice(n))

incoming: ----a--------b-------c----------d--------------------e-------->
list:  -------[a]----[a,b]----[a,b,c]----[a,b,c,d]--------->
selected object:  ---------------------------------c------->

new stream of list:                           ------[c,d]-----[c,d,e]--->

选中对象时获取不到列表流的最后一个值,,,为了更好地理解，做了一个弹珠图，

selectedObject$ =  new BehaviorSubject(0);
incomingObjects$ = new Subject();

list$ = incomingObjects$.pipe(
          scan((acc, val) => {
            acc.push(val);
            return acc;
          }, [])
        )
newList$ = selectedObject$.pipe(
            withLastFrom(list$),
          switchMap(([index,list])=> incomingObjects$.pipe(
             scan((acc, val) => {
              acc.push(val);
              return acc;
             }, list.slice(index))
          ))
        )

Answer 1

我与 scan 运算符一起使用的一个常见模式是传递 reducer 函数而不是要扫描的值，以便可以在更新操作中使用当前值。在这种情况下，您可以使用 merge 运算符链接这两个 observable，并将它们的值映射到适当的函数 - 添加到列表，或在选择后对列表进行切片。

// these are just timers for demonstration, any observable should be fine.
const incoming$ = timer(1000, 1000).pipe(map(x => String.fromCharCode(x + 65)), take(10));
const selected$ = timer(3000, 3000).pipe(map(x => String.fromCharCode(x * 2 + 66)), take(2));

merge(
  incoming$.pipe(map(x => (s) => [...s, x])), // append to list
  selected$.pipe(map(x => (s) => { // slice list starting from selection
    const index = s.indexOf(x);
    return (index !== -1) ? s.slice(index) : s;
  }))
).pipe(
  scan((list, reducer) => reducer(list), []) // run reducer
).subscribe(x => console.log(x)); // display list state as demonstration.