仅当 Rxjs 轮询上的某些属性更改时才调度 NgRx 操作答案

【问题标题】：Dispatch NgRx actions only when some properties changes on Rxjs Polling仅当 Rxjs 轮询上的某些属性更改时才调度 NgRx 操作
【发布时间】：2021-02-23 09:47:44
【问题描述】：

我有一个像这样的 Rxjs 轮询效果：

updateProductData$ = createEffect(() =>
  this.actions$.pipe(
    ofType(fromActions.loadProduct),
    switchMap(_) =>
      this.http.get('endpoint').pipe(
        delay(1000),
        repeat(),
        switchMap((data) => [
          fromActions.updateFoo({foo: data.foo}),
          fromActions.updateBar({bar: data.bar}),
        ])
      )
    )
  );

仅当data.foo 或data.bar 分别更改时，我如何才能调度updateFoo 和updateBar？

我可以通过使用distinctUntilChanged 来改进这一点，这样如果data.stuff 发生更改，这些操作将不会触发，但是，当任一操作发生更改时，这两个操作仍会分派。

...
     repeat(),
     distinctUntileChanged((prev, curr) => prev.foo === curr.foo && prev.bar === curr.bar) // works but it fires both actions when either one changes
     switchMap((data) => [
       fromActions.updateFoo({foo: data.foo}),
       fromActions.updateBar({bar: data.bar}),
     ])

我想在data.foo 更改时调度updateFoo，在data.bar 更改时调度updateBar，因为我知道data 有许多其他属性会随着时间的推移而更改。

【问题讨论】：

标签： angular rxjs ngrx

【解决方案1】：

我认为这可能是一种方法：

updateProductData$ = createEffect(() =>
  this.actions$.pipe(
    ofType(fromActions.loadProduct),
    switchMap(_) =>
      this.http.get('endpoint').pipe(
        delay(1000),
        repeat(),
        
        multicast(
          new Subject(),
          source => merge(
            // concerning `foo`
            source.pipe(
              distinctUntilChanged((prev, crt) => prev.foo === crt.foo),
              map(data => fromActions.updateFoo({foo: data.foo})),
            ),

            // concerning `bar`
            source.pipe(
              distinctUntilChanged((prev, crt) => prev.bar === crt.bar),
              map(data => fromActions.updateBar({bar: data.bar})),
            ),
          )
        )
      )
    )
  );

multicast 的第二个参数中的 source 是已声明为第一个参数的 Subject 的实例。通过使用multicast，我们可以将问题划分为另外两个较小的问题，而无需冗余订阅源（这就是使用主题的原因）。

【讨论】：

效果似乎没有返回任何动作，VScode 抱怨这一点。我错过了什么吗？
@GreatHawkeye 我不明白为什么它不会返回。除了类型错误，还有其他错误吗？也许您可以添加一些与此相关的代码？
你是对的，这是一个错字！然而，这两个动作在开始时都会触发一次，然后它会正常工作：轮询一直持续到 foo 或 bar 发生变化，然后它会调度动作。知道为什么动作一开始就会触发吗？这显然是因为一开始 prev.foo === curr.foo 是真的，但为什么呢？
只有一句话：multicast(new Subject(), selector) == publish(selector)
ncjamieson.com/calling-publish-with-a-selector

【解决方案2】：

您可以尝试使用pairwise 运算符来获取先前的状态并显式检查先前的值。我使用startWith(null) 来触发HTTP 请求中第一个值的发射。没有它，它不会在第二次调用之前开始发射

stop$ = new Subject<any>();
poll$ = timer(0, 1000).pipe(
  startWith(null),          // <-- emit `[null, data]` for first emission
  switchMap(this.http.get('endpoint')),
  pairwise(),
  map((data) => ({
    foo: {prev: data[0].foo, current: data[1].foo},
    bar: {prev: data[0].bar, current: data[1].bar}
  })),
  takeUntil(stop$)
);

updateProductData$ = createEffect(() =>
  this.actions$.pipe(
    ofType(fromActions.loadProduct),
    switchMap(_ =>
      poll$.pipe(
        switchMap((data) => {
          let result = [];

          if (data.foo.prev && data.foo.prev !== data.foo.current) 
            result.push(fromActions.updateFoo({foo: data.foo}))
          if (data.bar.prev && data.bar.prev !== data.bar.current) 
            result.push(fromActions.updateBar({bar: data.bar}))

          return result;
        })
      )
  )
);

注意：我没有使用过 NgRX，这是未经测试的代码。

【讨论】：