RXJS 管道行为主题

Question

我有一个默认为空对象的流。随着时间的推移，这个对象的键被填满。

  const RXSubject = new BehaviorSubject({});

  RXSubject.pipe(
    filter((frame): frame is InstDecodedFrame => frame.type === FrameType.INST),
    scan<InstDecodedFrame, InstantDataDictionnary>(
      (acc, frame) => ({ ...acc, ...frame.dataList }),
      {},
    ),
  );

现在我在应用程序的某些部分订阅过滤器，但如果我在其他地方订阅并且最后一个值没有触发过滤条件。我的新 observable 什么也得不到。

有什么方法可以从管道的任何订阅者的管道中获取最新的“有效”值？

谢谢

Answer 1

您可以在 filter() 之后使用 shareReplay(1) 并订阅该 observable：

const obs$ = RXSubject.pipe(
    filter((frame): frame is InstDecodedFrame => frame.type === FrameType.INST),
    scan<InstDecodedFrame, InstantDataDictionnary>(
      (acc, frame) => ({ ...acc, ...frame.dataList }),
      {},
    ),
    shareReplay(1),
  );

然后您将订阅obs$ 而不是RXSubject。