寻找满足条件直接发射值、否则执行后续操作的RXJS Operator

RxJS没有内置完全匹配你描述的操作符，但可以通过两种方式实现需求：

1. 用partition+merge拆分流（推荐） #

这种方式逻辑清晰，直接将源流拆分为满足条件和不满足条件的两个分支，分别处理后合并：

import { of, partition, merge, map } from 'rxjs';

const source$ = of(1, 2, 3, 4, 5);
// 拆分流：满足条件的直接发射，不满足的走后续操作
const [match$, noMatch$] = partition(source$, (v) => v <= 4);

merge(
  match$, // 满足条件，直接发射原数值
  noMatch$.pipe(map(v => v + 100)) // 不满足条件，执行map处理
).subscribe(v => console.log(v));
// 输出结果：1, 2, 3, 4, 105

2. 自定义操作符 #

如果你希望保持类似示例中的链式调用风格，可以自定义一个操作符，将后续的处理逻辑作为参数传入：

import { of, Observable, OperatorFunction } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

// 自定义操作符：满足条件则直接发射，否则执行传入的操作逻辑
function skipRestWhen<T>(condition: (val: T) => boolean, elseOperators: OperatorFunction<T, T>): OperatorFunction<T, T> {
  return mergeMap(val => {
    if (condition(val)) {
      return of(val);
    }
    return of(val).pipe(elseOperators);
  });
}

// 使用示例
const source$ = of(1, 2, 3, 4, 5);
source$
  .pipe(
    skipRestWhen(
      (v) => v <= 4,
      map(v => v + 100) // 不满足条件时执行的操作
    )
  )
  .subscribe(v => console.log(v));
// 输出结果：1, 2, 3, 4, 105

补充说明 #

你示例中的链式写法（将目标操作符放在map之前）无法直接实现，因为RxJS的pipe操作符是链式执行的，上游的每个值都会依次经过所有后续操作符。必须通过分支流的方式，让满足条件的值跳过后续处理逻辑。

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