防止一个主题的多个订阅者触发 RxJS 中每个订阅者的部分主题管道

Question

意图

我正在尝试尽可能地使用完全反应模式来实现 CRUD 屏幕（没有手动订阅，最好没有

tap

操作员，等等......）。

说明

我有两个异步管道订阅了模板中的

inventory$

，还有一个按钮发送一个描述创建、更新或删除操作的对象到

crudActionSubject

.

performRemoteCRUDAction

根据传入的操作进行 HTTP 调用，并且

performLocalCRUDAction

更新存储在

scan

的累加器中的本地状态。

  inventory$ = 
  
      this.http.get<Ingredient[]>("http://localhost:3000/ingredients").pipe(
        shareReplay(1),
        concatMap(ingredients => 
          merge(
            of(ingredients),  // initial ingredients
            this.crudActionSubject.pipe(  // this pipe runs for each subscriber -- but it should run once for each action sent to the subject
              concatMap((action) => this.performRemoteCRUDAction(action)),
              scan((ingredients, action) => this.performLocalCRUDAction(ingredients, action), ingredients)
            )
          ))
      );

问题

我的问题是，由于有两个异步管道，CRUD 操作两次触发 CRUD 操作管道——也就是说，我有两个 POST 请求、两个 PUT 请求等……

问题

我应该如何构造我的管道以避免这种情况？我尝试搜索 RxJS CRUD 示例，但它们中的大多数在某些时候使用带有/或副作用的命令式样式（例如手动订阅、

tap

运算符……）。

为了简洁起见，我省略了其他方法。如果需要，我会更新描述。

当前解决方案

在 IIFE 中构造一个中介主题并订阅管道解决了这个问题，但它是命令式的并且本质上是构造函数逻辑，如果可能的话，我很乐意看到一种更具反应性的方法。

  inventory$ = (() => {
  
      const o = // pipeline as seen above

      const s = new BehaviorSubject<Ingredient[]>([]);
      o.subscribe(s);

      return s;

  })();

Answer 1

这就是 shareReplay 的用途

inventory$ = 

  this.http.get<Ingredient[]>("http://localhost:3000/ingredients").pipe(
    shareReplay(1),
    concatMap(ingredients => 
      merge(
        of(ingredients),  // initial ingredients
        this.crudActionSubject.pipe(  // this pipe runs for each subscriber -- but it should run once for each action sent to the subject
          concatMap((action) => this.performRemoteCRUDAction(action)),
          scan((ingredients, action) => this.performLocalCRUDAction(ingredients, action), ingredients)
        )
      )),
    shareReplay(1)
  )