RxJS 5.5+中如何结合pipe()使用multicast()操作符？

这个问题我之前也碰到过，确实是RxJS 5.5+引入pipeable操作符后类型系统的变化导致的——链式调用时的multicast()返回的是ConnectableObservable，但pipe()里使用的pipeable版本multicast()返回的是普通Observable类型，所以直接调用.connect()会触发TypeScript的类型错误，虽然代码能运行，但类型检查不通过。

下面给你几个可行的解决方案：

方案一：类型断言为ConnectableObservable #

我们可以手动把pipe后的Observable断言成ConnectableObservable类型，这样就能正常调用connect()了，记得先导入对应的类型：

import { interval, Subject, ConnectableObservable } from 'rxjs';
import { filter, take, multicast } from 'rxjs/operators';

const even$ = new Subject();
// 用类型断言转换类型
const connectedObservable$ = interval(500)
  .pipe(
    filter(count => count % 2 === 0),
    take(5),
    multicast(even$)
  ) as ConnectableObservable<number>;

// 现在可以正常调用connect()了
const connectedSub = connectedObservable$.connect();
even$.subscribe(value => console.log(value));

方案二：使用connectable操作符（RxJS 6+推荐） #

如果你的项目已经升级到RxJS 6或更高版本，更推荐使用专门的connectable()操作符，它会直接返回ConnectableObservable，不需要额外的类型断言，用法更清晰：

import { interval, Subject, connectable } from 'rxjs';
import { filter, take } from 'rxjs/operators';

const even$ = new Subject();
const connectedObservable$ = connectable(
  interval(500).pipe(
    filter(count => count % 2 === 0),
    take(5)
  ),
  { connector: () => even$ }
);

const connectedSub = connectedObservable$.connect();
even$.subscribe(value => console.log(value));

额外补充：自动管理连接（无需手动调用connect） #

如果你不需要手动控制连接的时机，只是想实现多订阅共享源的效果，可以配合refCount()操作符，它会在第一个订阅者出现时自动调用connect()，最后一个订阅者取消时自动断开连接：

import { interval, Subject } from 'rxjs';
import { filter, take, multicast, refCount } from 'rxjs/operators';

const even$ = new Subject();
const sharedObservable$ = interval(500)
  .pipe(
    filter(count => count % 2 === 0),
    take(5),
    multicast(even$),
    refCount()
  );

// 第一个订阅触发自动连接
sharedObservable$.subscribe(value => console.log(value));

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