RXJS累积

Question

我正在从蓝牙设备接收历史记录帧，我想重新计算每个帧的历史记录。

我可以拥有这样的历史对象：

{
  value1: [
    { date: new Date(), value: 45}
    { date: new Date(), value: 40}
  ],
  value2: [
    { date: new Date(), value: true}
    { date: new Date(), value: false}
  ]
}

现在有时我会收到新的数据作为例子

{
  value1: [
    { date: new Date(), value: 60}
  ],
}

我想要得到

{
  value1: [
    { date: new Date(), value: 60}
    { date: new Date(), value: 45}
    { date: new Date(), value: 40}
  ],
  value2: [
    { date: new Date(), value: true}
    { date: new Date(), value: false}
  ]
}

我已经开始写这段代码了，但是我不确定是否有一个操作符已经…了

  RX.pipe(
        filter(frame => FrameTypes[frame.FrameType] === FrameTypes.HIST),
        scan((acc, val) => {
          // Merge data by hand (not done already)
        }, historyFromDatabase),
      ).subscribe(fullHistory => {
        console.log(fullHistory);
        // Update history to only keep values from the last two weeks
        const newHistory = cleanOldData(2, history)
        // Now save new history to database with previous and new values.
        saveHistory(newHistory)
      });
    }

你有没有更好的想法让我实现这一点？

向Andréas致敬

Answer 1

如果您的Observable是有限的(可以在here中找到解释)，您可以使用groupBy运算符按属性进行分组。

例如：

source$
  .pipe(
    mergeMap(history => Object.entries(history)),
    groupBy(([key]) => key),
    mergeMap(group => {
      return zip(
        of(group.key),
        group.pipe(
          mergeMap(([_, values]) => values),
          toArray()
        )
      );
    }),
    toArray(),
    map(entries => Object.fromEntries(entries))
  )
  .subscribe(fullHistory => console.log(fullHistory));

其中source$是你的历史记录，完整的源代码可以在this stackblitz中找到。

Answer 2

将您的扫描功能更新为

(acc, val) => {
  // Merge data by hand (not done already)
  Object.keys(val).forEach(key => {
    if (acc[key]) {
      acc[key] = [ ...val[key], ...acc[key] ];
    } else {
      acc[key] = val[key];
    }
  });
  return acc;
}