Angular RxJS，将新添加的项目推送到现有的 Observable Array

Question

我有2个服务。 RoomService -> 返回现有房间的列表 NotificationService -> 添加新房间时返回服务器发送事件（SSE）。

我想做的是

• 获取所有现有房间
• 等待SSE创建房间
• 收到 SSE 后，将这个新房间添加到现有房间列表中。

这就是我想要做的

export class RoomListComponent {
  rooms$: Observable<Room[]> = this.roomService
    .getRooms()
    .pipe(
      mergeMap((rooms) =>
        this.notificationService
          .getServerSentEvent<Room>()
          .pipe(concatMap((room) => [...rooms, room]))
      )
    );

  constructor(
    private notificationService: NotificationService,
    private roomService: RoomService
  ) {}
}

但这会导致错误

Type 'Observable<Room>' is not assignable to type 'Observable<Room[]>'. Type 'Room' is missing the following properties from type 'Room[]': length, pop, push, concat, and 29 more.ts(2322)

我做错了什么？

Answer 1

试试这个：

...    
rooms$: Observable<Room[]> = combineLatest([
  this.roomService.getRooms(),
  this.notificationService.getServerSentEvent<Room>()
]).pipe(
  map(x => [...x[0], x[1]])
);
...

Answer 2

使用 scan 来累积从您的

notificationService

发出的所有新房间

const initialRooms$ = roomService.getRooms();

const allNewRooms$ = notificationService.getServerSentEvent().pipe(
  scan((acc, curr) => [...acc, curr], []),
  startWith([])
);

// all rooms
combineLatest([initialRooms$, allNewRooms$])
  .pipe(map(([initialRooms, allNewRooms]) => [...initialRooms, ...allNewRooms]))
  .subscribe((arr) => console.log(JSON.stringify(arr)));

堆栈闪电战