Rx (JS)：如何在 flatMap 流中更新订阅

Question

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

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

我每次都尝试对合并的消息$进行平面映射，但遇到的问题是旧消息$s（在以前的 states$ 值中）被多次订阅。

我该如何解决这个问题？

let allMessages$ = state$.flatMap(s => { return Observable.merge(s.messages$s) } ) allMessages$.subscribe((x)=>{ console.log('message', x) // message from single message$ appear multiple times })

问题是，在 state$ 更新（推送项目）后，旧的会被多次订阅。 state$ --s(1)---------s(2)---- message$s[0]. --m1----m2-----------m4-- message$s[1] ---------------m3-------- allMessages$ --m1----m2-----m3----m4 m1 m4

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

我想要的是： state$ --s(1)---------s(2)----- message$s[0] --m1----m2-----------m4-- message$s[1] ---------------m3-------- allMessages$ --m1----m2-----m3----m4

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

Answer 1

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

• 发射state1
  • 订阅typing$
• 发射state2
  • 订阅typing$
  • 订阅typing2$

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

• 发射state1
  • 订阅typing$
• 发射state2
  • 'unsubscription'（甚至是英语）来自flatMapLatest 中发出的先前流，即Rx.Observable.merge(state.items) 即typing$
  • 订阅typing$
  • 订阅typing2$

因此，请尝试将 flatMap 替换为 flatMapLatest，如果这样可以解决问题，请告诉我。

解决此问题的另一种方法也可以使用状态更改流而不是整个状态（redux 为反应所做的那种）。