switchMap内部动态数组的CombineLatest不断取消订阅和重新订阅

Question

我至少有两个按钮想要动态监听点击。listeningArray$将发出一个ar我需要监听的按钮 # 的数组 ( )。当有人单击我正在收听的这些按钮之一时，我需要控制台记录所单击的按钮，并记录某个时间间隔的值。

如果ar从 到[1,2]，[1]我们需要停止监听按钮 #2 的点击。因此，需要删除 DOM 单击事件2，这应该会触发.finally()操作符。但是对于1，我们应该保持订阅状态，并且 中的代码.finally()不应该运行，因为没有任何内容被取消订阅。

const obj$ = {};

Rx.Observable.combineLatest(
  Rx.Observable.interval(2000),
  listeningArray$ // Will randomly emit either [1] or [1,2]
)
.switchMap(([x, ar]) => {
    const observables = [];

    ar.forEach(n => {
        let nEl = document.getElementById('el'+n);

        obj$[n] = obj$[n] || Rx.Observable.fromEvent(nEl, 'click')
          .map(()=>{
            console.log(' el' + n);
          })
          .finally(() => {
            console.log(' FINALLY_' + n);
          });

        observables.push(obj$[n]);
    })

    return Rx.Observable.combineLatest(...observables);
})
.subscribe()

但发生的情况是，每次间隔发出一个值时，DOM 事件都会被删除，然后立即再次添加，并且 .finally 运算符内的代码会运行 for1 和 2。

这真的让我很沮丧。我缺少什么？

这是一个有点复杂的情况，所以我创建了这个：https://jsfiddle.net/mfp22/xtca98vx/7/

Answer 1

我实际上非常接近，但我误解了要点switchMap。

switchMap旨在取消订阅每当从上面发出新值时它返回的可观察值。这就是为什么当需要发出新请求时，可以使用它来取消旧的待处理 Http 请求。

我遇到的问题是可以预料到的。switchMap在订阅当前可观察对象之前，将取消订阅先前返回的可观察对象。正如我在问题中解释的那样，这是不可接受的。这是不可接受的原因是，在我的实际项目中，可fromEvent观察量正在侦听 Firebasechild_added事件，因此当这些冷可观察量从没有订阅者变成有 1 个订阅者时，Firebase 随后会为每个已经存在的子项触发该事件，以及供将来添加。

我玩了mergeMap一段时间，但是手动取消订阅之前返回的可观察量确实很困难并且有很多问题。

switchMap因此，我在执行其过程时为内部可观察量添加了一个订阅者unsubscribe from old => subscribe to new，以便始终有一个订阅者。我曾经takeUntil(Observable.timer(0))确保订阅者不会增加并导致内存泄漏。

可能有更好的解决方案，但这是我发现的最好的解决方案。

const obj$ = {};

Rx.Observable.combineLatest(
  Rx.Observable.interval(2000),
  listeningArray$ // Will randomly emit either [1] or [1,2]
)
.switchMap(([x, ar]) => {
  const observables = [];

  ar.forEach(n => {
    let nEl = document.getElementById('el'+n);

    obj$[n] = obj$[n] || Rx.Observable.fromEvent(nEl, 'click')
      .map(()=>{
        console.log(' el' + n);
      })
      .finally(() => {
        console.log(' FINALLY_' + n);
      })
      .share();
    
    obj$[n].takeUntil(Rx.Observable.timer(0))
      .subscribe();
    observables.push(obj$[n]);
  })

  return Rx.Observable.combineLatest(...observables);
})
.subscribe()

我还必须添加该.share()方法。无论如何我都需要它。我使用这种模式让一些 Angular 组件声明它们需要什么数据，忽略其他组件可能需要什么，以实现更好的关注点分离。因此，多个组件可以订阅相同的 Firebase observables，但.share()操作员确保来自 Firebase 的每条消息仅处理一次（我将每个消息分派到 Redux 存储）。

工作解决方案：https://jsfiddle.net/mfp22/xtca98vx/8/