Rx 中的 Array Join ish 运算符答案

【问题标题】：Array Join ish operator in RxRx 中的 Array Join ish 运算符
【发布时间】：2018-12-07 19:40:41
【问题描述】：

我有一个Observable<EntityId[]>，并且可以为每个 EntityId 获得一个Observable<Entity>。我想将其转换为Observable<Entity[]>。下面以简单的方式解决它：

idsStream
    .Select(ids => ids
        .Select(id => StreamEntity(id))
        .CombineLatest())
    .Switch()

问题是每次 idsStream 发出一些新的 id 时，每个实体订阅都会被关闭，而新的订阅也会被打开。我只想关闭不再包含在EntityId[] 中的 Observable，并为添加的 id 打开新的流。

在我看来，应该有一个运营商，但我找不到任何东西。有谁知道这样的运算符，或者我应该自己编写/实现。如果是这样，以下似乎是一个想法

// Only select for new elements and reuses former values
SelectAndKeep: Observable<TIn[]> -> (TIn -> TOut) -> Observable<TOut[]>

// Only subscribes/unsubscribe to new/removed observables
SwitchKeepAndCombineLatest: Observable<Observable<T>[]> -> Observable<T[]>

Observable<EntityId[]> idsStream

Observable<Observable<Entity>[]> entitiesStreamStream = 
    idsStream.SelectAndKeep(id => StreamEntity(id))

Observable<Entity[]> entitiesStream = entitiesStreamStream
    .SwitchKeepAndCombineLatest()

对这种情况有什么想法或经验吗？它似乎很通用，因此应该存在一个好的解决方案......

【问题讨论】：

嗨 Andreas，欢迎来到 stackoverflow，以便为您提供必须优雅的解决方案。请考虑发布可执行代码 whitch 将作为 pruposal 的基础。在stackblitz 是正确的地方。无需提供完整的代码，只需提供有针对性的问题即可。

标签： rxjs reactive

【解决方案1】：

根据我对您问题的理解，我建议您执行此操作：

interface Entity {
  id: number;
  title: string;
}
const entities: Entity[] = [{id: 1, title: 'foo1'}, {id: 2, title: 'foo2'}, {id: 3, title: 'foo3'}];

const entities$ = of(entities); // Observable of Entity[]

const ids$ = of([1,2], [1,3]); // Observable of ids.

ids$.pipe(mergeMap(ids => { 
  // Transform collection of id to collection of entity
  return entities$.pipe(map((entities) => {
      // from whole collection of entity, let keep what is present on ids.
      return entities.filter(entity => ids.findIndex(id => entity.id === id) !== -1);
  }));
})).subscribe(console.log);

entities$ Observable 应该是 Hot observable（如 BehaviorSubject），以便始终接收您商店中可用的最后实体。 mergeMap 并行合并并将 ids$ 流上的每个新数据映射到由原始 ids 集合过滤的实体集合。

live example

【讨论】：