如何实现Observable发射值覆盖：override$优先于default$的延迟判断逻辑

我有两个经常在相近时间发射值的Observable：default$和override$。需求如下：

• 每当default$发射时，等待50ms查看override$是否也发射：若override$在50ms内发射，则使用其值；否则使用default$的值。
• 即使default$从不发射，override$发射时也要执行操作。

测试用例代码：

const default$ = timer(0, 1000).pipe(map(() => 'default'), take(10));
const override$ = timer(49, 2000).pipe(map(() => 'override'), take(5));

尝试过用debounceTime()实现，但无法满足override$优先处理的逻辑：

const result = merge(default$, override$).pipe(debounceTime(50)).subscribe((val) => {
    console.log(val);
});

想知道如何用switchMap()实现需求。

可以通过合并两个流分支的方式实现需求：一个分支处理override$的直接发射，另一个分支处理default$的延迟判断逻辑，结合switchMap和takeUntil操作符完成。

实现代码 #

import { timer, merge } from 'rxjs';
import { map, take, switchMap, takeUntil } from 'rxjs/operators';

const default$ = timer(0, 1000).pipe(map(() => 'default'), take(10));
const override$ = timer(49, 2000).pipe(map(() => 'override'), take(5));

const result$ = merge(
  // 分支1：直接响应override$的发射，满足"无default$时也处理override$"的需求
  override$,
  // 分支2：处理default$的延迟判断逻辑
  default$.pipe(
    switchMap(defaultVal => 
      // 启动50ms定时器，准备输出default值
      timer(50).pipe(
        mapTo(defaultVal),
        // 若期间override$发射，立即终止定时器流，不输出default值
        takeUntil(override$)
      )
    )
  )
);

result$.subscribe(val => console.log(val));

逻辑说明 #

1. override$分支：直接订阅override$，只要它发射值就立即输出，完全满足“无default$时也要响应override$”的要求。
2. default$分支：
   • 每次default$发射值后，通过switchMap创建一个新的定时器流（延迟50ms后输出default值）。
   • 用takeUntil(override$)监听override$：如果50ms内override$发射，就立刻终止定时器流，不会输出default值；如果50ms内没有override$发射，定时器到期后就输出default值。

为什么debounceTime不合适 #

merge(default$, override$).pipe(debounceTime(50))的逻辑是等待最后一次发射后的50ms才输出值：

• 比如default$在0ms发射，override$在49ms发射，debounceTime会从49ms开始计时50ms，到99ms才输出override，不符合“override$优先且立即响应”的需求；
• 同时它无法区分是default$还是override$触发的最终输出，逻辑完全不符合需求。

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