RXJS中如何重映射可观察对象为计时起始值且不中断原数据流

原生操作符最简实现 #

无需自定义通用操作符，直接用RxJS内置能力即可满足需求：

实现返回起始时间的startTimer操作符 #

import { defer, ignoreElements, concatWith, of } from 'rxjs';

function startTimer() {
  return source$ => defer(() => {
    const startTime = Date.now();
    return source$.pipe(
      ignoreElements(), // 忽略源流所有next输出，仅保留完成、错误通知
      concatWith(of(startTime)) // 源流完成后发射计时起始值
    )
  })
}

调用方式和你预期的完全一致：

obs.pipe(take(1000), startTimer())
   .subscribe(start => {
       // 展示流式传输1000个值耗费的总时长
       const duration = Date.now() - start;
       console.log(duration); 
   });

直接输出耗时的简化版本 #

如果不需要自行计算时长，可以直接在操作符内完成计算，下游直接拿到最终耗时：

import { defer, ignoreElements, concatWith, of } from 'rxjs';

function measureDuration() {
  return source$ => defer(() => {
    const startTime = Date.now();
    return source$.pipe(
      ignoreElements(),
      concatWith(of(Date.now() - startTime))
    )
  })
}

// 调用示例
obs.pipe(take(1000), measureDuration())
   .subscribe(duration => console.log(duration));

自定义drain操作符的补充说明 #

你自己实现的通用drain操作符是合理的，适合有多处「排空源流后发射自定义值」需求的场景，可以作为通用工具函数复用。如果仅当前计时场景使用，原生操作符组合的方案更轻量化，无需额外维护自定义工具。 #

内容的提问来源于stack exchange，提问作者vitaly-t