许多过滤订阅的可观察流答案

【问题标题】：Observable Stream to Many Filtered Subscriptions许多过滤订阅的可观察流
【发布时间】：2020-04-19 05:30:27
【问题描述】：

我正在为一个名为 multisubscribe 的 API 编写服务，该服务接受一组变量并开始通过 HTTP 发送数据流。数据是 JSON 有效负载，在更改时发送可变数据。

{device: "OEE", variable: "Availability", data: [12]}

当取消订阅 HTTP 请求被中止时，我已经设法解析流并创建一个发出这些 JSON 有效负载的 observable。完美的！ You can see the code on GitHub.

现在我正在尝试将 observable 抽象到另一个名为 MultisubscribeStore 的服务中。我基本上希望能够添加、删除、编辑变量，并且每次取消订阅、重新订阅时都会在服务中自动发生。这样用户就不必考虑单个流。

我尝试创建一个单独的主题并在该主题上发布，但我无法确定主题何时完成，所以我无法取消订阅。

我还希望能够将流过滤成许多流并将它们传递给子组件。例如，我有 5 个仪表组件，它们分别代表不同的数据。我怎样才能向他们每个人发送一个只会发出他们感兴趣的值的可观察对象。我觉得使用过滤器创建多个新的可观察对象性能很差。

const subscription1 = multiSubscribe$.pipe(
  filter((data) => data.variable === 'Availability')
);

const subscription2 = multiSubscribe$.pipe(
  filter((data) => data.variable === 'Quality')
);

const subscription3 = multiSubscribe$.pipe(
  filter((data) => data.variable === 'Performance')
);

也许我想要的已经内置在 RXJS 中了。我尝试使用共享和多播，但无济于事。

【问题讨论】：

标签： angular typescript rxjs observable angular8

【解决方案1】：

由于您只有一个流，因此您需要过滤事件并仅保留特定仪表在某个时候需要的事件。

在ngrx中可以找到类似的场景，我们有一个动作流，效果正在监听这个流并只过滤他们感兴趣的动作（使用ofType("MyAction")运算符，它只是在.filter(action => action.type === "MyAction")下做引擎盖）。恐怕没有别的办法了

【讨论】：