rxjs groupBy 并将每个发出的对象与最新的其他组进行比较

Question

是否可以使用 rxjs 对对象属性进行分组，并将一组的新发出事件与其他组的最新值进行比较？在此之后将所有流合并为一个。

例如，我有 5 个房间，每个房间都有一盏灯。发射器随机打开或关闭其中一个房间的灯。我想知道管道中所有房间的灯何时关闭或打开，以及何时将属性 allLightsOff 添加到发射的对象。

const source = interval(1000).pipe(
  map((e) => {
    return {
      iteration: e,
      room: Math.floor(Math.random()*5),
      lightOn: Math.round(Math.random()),
      allLightsOff: null
    };
  }),
  groupBy((o) => o.room)
  // how to compare one room with all others?
  // how to merge alls groups to a single stream together again?
);
const subscribe = source.subscribe((o) => console.log(o));

Answer 1

我认为您可以使用ReplaySubject(1) 和switchMap + combileLatest 来执行以下操作：

const source = interval(1000).pipe(
  map((e) => {
    return {
      iteration: e,
      room: Math.floor(Math.random()*5),
      lightOn: Math.round(Math.random()),
      allLightsOff: null
    };
  }),
  groupBy((o) => o.room, undefined, undefined, () => new ReplaySubject(1)),
  scan((allGroups, group) => [...allGroups, group], []),
  switchMap(groups => combineLatest(groups)),
  map((rooms: any[]) => {
    if (rooms.every(room => room.lightOn)) {
      console.log(rooms);
      return 'all on';
    } else if (rooms.every(room => !room.lightOn)) {
      console.log(rooms);
      return 'all off';
    }
    return 'some on, some off';
  }),
);

const subscribe = source.subscribe((o) => console.log(o));

每个新组是ReplaySubject(1)，它会重放其最后一个值，因此当添加一个新组时，它将由scan 累加，然后combileLatest 将订阅新列表。

现场演示：https://stackblitz.com/edit/rxjs-u9w73k?devtoolsheight=60