javascript - 在 flatMapLatest 中重新使用订阅

Question

我有以下 RxJS 定义:

onIds
    .flatMapLatest((ids) => Rx.Observable
      .from(ids)
      .flatMap((id) => onUpdated(id)))

效果很好。然而，onUpdated 是一个昂贵的重启订阅(即它有一个套接字连接)。因此，如果有一种方法可以重用以前的 ids 实例的订阅，我会更喜欢什么，但是我不确定如何以务实的方式实现这一点......

编辑:

我能想到的最好的是:

function cached(factory) {
  return (id) => {
    const cache = factory._cached_cache || (factory._cached_cache = new Map())
    const x = cache.get(id)
    return x ? x : Rx.Observable.defer(() => {
        const x = factory(id)
          .finally(() => cache.delete(id))
          .share()
        cache.set(id, x)
        return x
      })
    }
}

onIds
    .flatMapLatest((ids) => Rx.Observable
      .from(ids)
      .flatMap(cached(onUpdatedCache))

Answer 1

发布更多关于 onUpdated 的信息会有所帮助.在不知道更多的情况下，我会说解决方案最有可能使用一些依赖注入(inject)技术将实时套接字连接作为参数传递给 onUpdated而不是让该方法创建自己的方法。

要么

Rx.Observable.using(
    () => new SocketConnection(...),
    socket => onIds.flatMapLatest(ids => Observable
        .from(ids)
        .flatMap(id => onUpdated(socket, id))));

或者进行一些封装......

Rx.Observable.using(
    () => new Updater(...),
    updater => onIds.flatMapLatest(ids => Observable
        .from(ids)
        .flatMap(id => updater.onUpdated(id))));

编辑:

根据您更新的描述，除了您的内存技巧之外，我能想到的唯一方法是对 id 的实际进入和退出进行建模。来自 ids大批。像这样:

onIds
    .startWith([])
    .pairwise()
    .flatMap([prev, curr] => {
        // generate list of "added" and "removed" ids
        const added = curr.filter(id => prev.indexOf(id) === -1);
        const removed = prev.filter(id => curr.indexOf(id) === -1);

        // only emit ids when they are added and later removed
        return added.concat(removed);
    })
    .groupByUntil(
        id => id, // key selector
        null, // no need for element selector
        ids => ids.skip(1)) // end the group when the id is seen 2nd time (e.g. removed)
    .map(group => group.key)
    .flatMap(id => doUpdate(id));