如何阻止已完成的hot Observable通过shareReplay(1)共享缓存值

你使用的默认shareReplay(1)等价于如下完整配置：

shareReplay({
  bufferSize: 1,
  refCount: false,
  resetOnComplete: false,
  resetOnError: false,
  resetOnRefCountZero: false
})

其中resetOnComplete: false意味着源流完成后缓存会永久保留，新订阅发起时会直接重放缓存的最后一个值，再发送完成通知。

方案1：RxJS 7+ 直接调整shareReplay配置 #

只需开启resetOnComplete参数，源流完成时会自动清空缓存，后续新订阅将不会收到任何历史缓存值：

// 替换原有 shareReplay(1) 即可
shareReplay({
  bufferSize: 1,
  resetOnComplete: true,
  refCount: false // 保持原shareReplay(1)的非引用计数行为，需要无订阅时自动销毁缓存则设为true
})

方案2：自定义share配置（RxJS 6.5+ 兼容） #

如果你的RxJS版本低于7，或者需要更灵活的多播控制，可以用share操作符搭配ReplaySubject实现相同效果：

import { share, ReplaySubject } from 'rxjs';

const yourMulticastObservable = originalHot$.pipe(
  share({
    connector: () => new ReplaySubject(1),
    resetOnComplete: true,
    resetOnError: false, // 发生错误时是否清空缓存，可按需调整
    resetOnRefCountZero: false // 无活跃订阅时是否清空缓存，可按需调整
  })
)

进阶：完成后新订阅不接收任何通知（含完成通知） #

如果你需要源流结束后新订阅完全收不到任何通知（包括默认的完成通知），可以追加一层状态判断：

import { defer, share, ReplaySubject, NEVER, tap } from 'rxjs';

let isSourceCompleted = false;
const yourMulticastObservable = defer(() => {
  // 源已完成时直接返回空流，不发送任何通知
  if (isSourceCompleted) return NEVER;
  return originalHot$.pipe(
    tap({ complete: () => isSourceCompleted = true }),
    share({
      connector: () => new ReplaySubject(1),
      resetOnComplete: true
    })
  )
})

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