javascript - Rxjs): how to update subscriptions in flatMap stream

Question

我有一个包含 messages$s 的 state$ 流，它是 messages$ 流的数组。 State$ 已更新并且新的 messages$ 出现。

我希望订阅者在一个流中处理来自所有消息$的消息，并且我希望该流仅包含正确的事件。

我每次都尝试使用 flatMap 合并 messages$，但是遇到了旧的 messages$(在之前的 states$ 值中)被多次订阅的问题。

如何解决这个问题？


让 allMessages$ = state$.flatMap(s => {
返回 Observable.merge(s.messages$s)
}
)
allMessages$.subscribe((x)=>{
console.log('消息', x)
//来自单个消息$的消息出现多次
})


问题是在 state$ 更新(推送项目)后，旧的状态会被多次订阅。
状态$ --s(1)---------s(2)----
消息$s[0]。 --m1----m2------------m4--
消息$s[1] ---------------m3--------
allMessages$ --m1----m2----m3----m4
米1米4


s(1) - 当状态有 1 个 message$ 时，s(2) 当添加第二个 message$ 时因此 allMessages$ 会触发来自 item1 的消息。

我想要的是:
状态$ --s(1)---------s(2)-----
消息$s[0] --m1----m2------------m4--
消息$s[1] ---------------m3--------
allMessages$ --m1----m2----m3----m4


此文件显示了简化的情况: http://jsfiddle.net/8jFJH/797/

Answer 1

根据您的简化情况，这些是订阅序列(您可以查看答案 here 以了解热可观察量与冷可观察量的解释以及对订阅流程的理解):

• 发射state1
  • 订阅typing$
• 发射状态2
  • 订阅typing$
  • 订阅typing2$

因为您使用flatMap，所以您同时拥有三个订阅。如果您使用 flatMapLatest 会发生以下情况:

• 发射state1
  • 订阅typing$
• 发射状态2
  • 来自 flatMapLatest 中发出的先前流的“取消订阅”(甚至是英语)，即 Rx.Observable.merge(state.items) 即 typing$
  • 订阅typing$
  • 订阅typing2$

因此，请尝试将 flatMap 替换为 flatMapLatest 并让我知道这是否可以解决问题。

解决这个问题的另一种方法也可以使用状态更改流，而不是整个状态(类似于 redux 为 React 所做的事情)。