RxJS合并Observable报错：You provided 'undefined' where a stream was expected

为什么最初会触发You provided 'undefined' where a stream was expected.错误？ #

你一开始的写法有个关键问题：refresh$只做了类型声明但没有初始化，默认是undefined。虽然你在TimerObservable的subscribe回调里给它赋值了，但subscribe是异步执行的——当你执行merge(this.refresh$, ...)这行代码时，定时器的回调还没触发，refresh$依然是undefined，这就导致merge收到了一个非Observable的值，直接抛出错误。

修改后的写法为什么能解决问题？ #

你后来用switchMap重构refresh$的创建逻辑，本质上是把refresh$从一个“待赋值的变量”变成了一个自包含的Observable流：

this.refresh$ = TimerObservable.create(0, (this.storageService.refRate * 500))
  .takeWhile(() => this.interval)
  .switchMap(() => this.funnelService.getStoryFunnels({}).map(response => [...response]));

• TimerObservable.create会按设定的间隔发射事件（第一个参数0表示立即触发一次）
• 每次定时器事件触发时，switchMap会调用funnelService.getStoryFunnels获取数据，并把返回的Observable的结果转发出来
• 这样refresh$从声明时就是一个合法的Observable，不会再出现undefined的情况，merge操作就能正常合并这个流了

补充说明 #

保留的合并代码逻辑是没问题的：

this.funnels$ = this.funnelService.getStoryFunnels({}).pipe(
  merge(
    this.refresh$, 
    funnelCreation$.pipe(
      filter(funnel => !!funnel),
      map(funnel => [funnel])
    )
  )
);

它会把初始获取的漏斗数据、定时刷新的漏斗数据，以及新创建的漏斗数据合并成一个流，完全符合你的业务需求。

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