角rxjs变化异步可观察到从web套接字

Question

嗨，我正在使用angular 7和rxjs async

在我的组件中，我使用ngFor与异步观察者

<item-comp [item]="item" *ngFor="let item of groupsService.selectedItems$ | async; ">

</item-comp>

在我的服务中，我有一个BehaviorSubject，当用户选择一个组时，它会被释放

  public groupSelected$: BehaviorSubject<any> = new BehaviorSubject(null);

这是selectedItems$可观察的：

public selectedItems$ = this.groupSelected$.pipe(
    switchMap((group: any) => {
        if (!group)
          return new EmptyObservable();

        return this.http.get('/api/'+ group)
          .pipe(
            map((res: any) => {
                return res.items;
              }
            )
          )
      }
    )
  )

这是可行的，但现在我需要能够更改特定项以响应websocket消息。我有一个websocket连接，用于处理更新项目的消息。有没有一种方法可以使用rxjs的反应式方法来实现这一点？

Answer 1

您可以为websocket更新创建更高阶的函数，然后将其与http请求链接起来

const onUpdate=(items)=>updateFromWebSocket.pipe(
   map(itemUpdates=>{........ return updatedItems}
   startWith(items)
)

selectedItems$.pipe(switchMap(items=>onUpdate(items)).subscribe()