如何在mergeMap内处理条件？避免返回of(null)触发订阅回调

问题说明 #

现有RxJS代码通过mergeMap处理Observable流，需求是仅当data.condition为true时，才触发subscribe回调中的doSomethingElse()。移除return of(null)后出现TypeScript类型错误：

Type 'Subject | undefined' is not assignable to type 'ObservableInput'.
Type 'undefined' is not assignable to type 'ObservableInput'.ts(2345)

原代码：

Obs1$.pipe(
  mergeMap((data) => {
    if (data.condition) {
      const generatedStuff = doSomethingFunction(data);
      return generatedStuff.Obs2$;
    }

    someCleanupAction();

    return of(null); // 希望移除该行，避免condition为false时触发doSomethingElse()
  })
).subscribe(() => {
  doSomethingElse();
})

解决方案 #

方案1：提前过滤不符合条件的数据流 #

使用filter操作符先过滤掉condition为false的数据流，后续仅处理符合条件的情况，直接避免进入不需要的分支，同时解决类型问题。

情况1：清理动作仅在condition为false时执行

import { tap, filter, mergeMap } from 'rxjs';

Obs1$.pipe(
  tap(data => {
    if (!data.condition) {
      someCleanupAction(); // 仅当condition为false时执行清理
    }
  }),
  filter(data => data.condition), // 过滤掉condition为false的数据流
  mergeMap((data) => {
    const generatedStuff = doSomethingFunction(data);
    return generatedStuff.Obs2$;
  })
).subscribe(() => {
  doSomethingElse();
});

情况2：清理动作在condition为true时执行

import { filter, mergeMap } from 'rxjs';

Obs1$.pipe(
  filter(data => data.condition),
  mergeMap((data) => {
    someCleanupAction();
    const generatedStuff = doSomethingFunction(data);
    return generatedStuff.Obs2$;
  })
).subscribe(() => {
  doSomethingElse();
});

方案2：返回EMPTY Observable替代of(null) #

使用RxJS的EMPTY常量，它是一个不发射任何值、直接完成的Observable，既满足TypeScript的类型要求，又不会触发subscribe的next回调。

import { EMPTY, mergeMap } from 'rxjs';

Obs1$.pipe(
  mergeMap((data) => {
    if (data.condition) {
      const generatedStuff = doSomethingFunction(data);
      return generatedStuff.Obs2$;
    }

    someCleanupAction();
    return EMPTY; // 不会触发subscribe的next回调
  })
).subscribe(() => {
  doSomethingElse();
});

方案对比 #

• 方案1更高效，直接过滤掉不需要处理的数据流，减少后续操作符的执行开销。
• 方案2更灵活，适合需要在mergeMap内部处理分支逻辑，同时控制是否触发回调的场景。

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