RxJS 5升级至6迁移问题：connect运算符替代方案咨询

在RxJS v6中，publishReplay作为pipeable操作符使用时，返回的是普通Observable类型，不再是v5中的ConnectableObservable，因此无法直接调用connect()方法。以下是两种保持原手动连接逻辑的等效实现方案：

方案一：使用connectable操作符（推荐，RxJS 6.5+支持） #

通过connectable操作符将普通Observable包装为可连接的Observable，配置中指定publishReplay作为连接器，之后即可调用connect()手动启动数据流。

import { from, of, connectable } from 'rxjs';
import { concatMap, publishReplay } from 'rxjs/operators';

const syncStatesObservable = connectable(
  from(intervals).pipe(
    concatMap(i =>
      fetchSyncState(i).pipe(
        concatMap(syncState => {
          if (syncState.version !== version) {
            logger.info(`Regenerate all entries: version ${version}`);
            return clearAndFetchSyncState(i);
          }
          return of(syncState);
        })
      )
    )
  ),
  { connector: () => publishReplay() }
);

// 手动启动连接，与原v5代码逻辑一致
syncStatesObservable.connect();

// 后续可在其他流中使用syncStatesObservable

方案二：直接创建ConnectableObservable #

如果你的RxJS版本低于6.5，可通过ConnectableObservable构造函数直接创建，配合publishReplay的底层实现：

import { from, of, ConnectableObservable } from 'rxjs';
import { concatMap, publishReplay } from 'rxjs/operators';

const source$ = from(intervals).pipe(
  concatMap(i =>
    fetchSyncState(i).pipe(
      concatMap(syncState => {
        if (syncState.version !== version) {
          logger.info(`Regenerate all entries: version ${version}`);
          return clearAndFetchSyncState(i);
        }
        return of(syncState);
      })
    )
  )
);

// 手动创建ConnectableObservable
const syncStatesObservable = new ConnectableObservable(source$, publishReplay());

// 启动连接
syncStatesObservable.connect();

补充：若无需手动控制连接 #

如果不需要手动调用connect()，而是希望订阅时自动连接、退订时自动清理，可直接使用shareReplay操作符替代，这是更常用的简化方案：

import { from, of } from 'rxjs';
import { concatMap, shareReplay } from 'rxjs/operators';

const syncStatesObservable = from(intervals).pipe(
  concatMap(i =>
    fetchSyncState(i).pipe(
      concatMap(syncState => {
        if (syncState.version !== version) {
          logger.info(`Regenerate all entries: version ${version}`);
          return clearAndFetchSyncState(i);
        }
        return of(syncState);
      })
    )
  ),
  shareReplay() // 替代publishReplay + connect，自动管理连接
);

// 首次订阅时自动启动数据流，无需手动调用connect

内容的提问来源于stack exchange，提问作者XeniaSis