在两个内部 observables 之间切换：switchMap 不会取消订阅之前的订阅

Question

我正在构建一个屏幕，其中包含两个 Ionic segments（每个分页实体类型一个，比如“小马”和“外星人”）和一个列表。对于它们中的每一个，我必须显示一个无限列表（不是真的），并带有无限滚动。

在我的模板中，我订阅了一个名为 entities$ 的 observable，如下所示：

模板

<ng-container *ngIf="(entities$ | async) as entities">
      <ng-container *ngIf="entities.length > 0; else noResult">
        <ion-col>
          <ion-list>
            <entities-item
              *ngFor="let entity of entities; trackBy: trackByFn"
              [entity]="entity">
            </entities-item>
          </ion-list>
          <ion-infinite-scroll threshold="100px" (ionInfinite)="loadMoreData($event)">
            <ion-infinite-scroll-content
              loadingSpinner="bubbles"
              [loadingText]="'common.actions.loadingMore' | translate">
            </ion-infinite-scroll-content>
          </ion-infinite-scroll>
        </ion-col>
      </ng-container>
    </ng-container>

在我的课堂上，我正在像这样构建我的可观察对象：

类

    const poneys$ = this.myService.entities$.pipe(
      map(entities => entities.filter(entity => entity.type === EntityType.PONEYS)),
      scan((acc, poneys) => [...acc, ...poneys]),
      tap(data => console.log('PONEYS')),
      shareReplay(1)
    );

    const aliens$ = this.myService.entities$.pipe(
      map(entities => entities.filter(entity => entity.type === EntityType.ALIENS)),
      scan((acc, aliens) => [...acc, ...aliens]),
      tap(data => console.log('ALIENS')),
      shareReplay(1)
    );

    this.entities$ = this.segment.ionChange.pipe(
      pluck('detail', 'value'),
      tap(segment => this.currentSegment = segment),
      startWith(this.currentSegment),
      switchMap(segment => segment === EntityType.PONEYS ? poneys$ : aliens$)
    );

其他详情

entities$ 来自 BehaviorSubject（asObservable 方法）。

我使用scan 运算符来保留我之前获取的实体集合并与新的列表发射相连接。

我使用shareReplay(1) 确保在我的两个可观察对象之间每次切换后都能获得最新的发出列表（我不想再次获取已经获取的数据）。

我的问题 在内部 observable 之间切换后，如果我滚动以获取更多数据，控制台会显示“ALIENS”和“PONEYS”，同时它应该显示当前订阅的 observable 的日志。 看来以前的订阅永远不会被switchMap取消订阅。

我是不是误会了什么？

Ty 提前！

Answer 1

switchMap 确实取消订阅。由于shareReplay，您会得到这种行为。默认情况下，当共享的 observable left 没有订阅者时，它不会取消订阅源。

如果你想从scan这里获取最新的累积值

const source = this.myService.entities$.pipe(
  map(entities => entities.filter(entity => entity.type === EntityType.X)),
  scan((acc, x) => [...acc, ...x])
)

那么有一个订阅在这个 observable 上运行，否则你会丢失累积的值。您可以在切换后使用shareReplay({ buffersize: 1, refCount: true }) 取消对内部源的订阅，但是如果您切换回来，则必须对source observable 建立新的订阅（即来自scan 的累积值将丢失） .所以在这种情况下使用shareReplay({ buffersize: 1, refCount: true }) 不会给你想要的效果（即与不使用shareReplay 相同）。

由于您想保留当前累积的实体，您需要保持对那些开放的订阅。您可以只为所有类型的累积实体创建一个流，而不是在不同的实体流之间来回切换。然后将此流与当前段组合并提取给定段的累积实体。

const accumulatedEntities$ = this.myService.entities$.pipe(
  scan((acc, entities) => groupBy(entities, "type", acc), {})
);

const segment$ = this.segment.ionChange.pipe(
  pluck('detail', 'value'),
  tap(segment => this.currentSegment = segment),
  startWith(this.currentSegment)
);

this.entities$ = combineLatest([accumulatedEntities$, segment$]).pipe(
  map(([entities, segment]) => entities[segment])
);

const groupBy = (data, key, initial) => {
  return data.reduce((storage, item) => {
    const group = item[key];
    storage[group] = storage[group] || [];
    storage[group].push(item);
    return storage;
  }, initial);
};

https://stackblitz.com/edit/rxjs-xhje7b?file=index.ts