寻找类似bufferWhen但仅保留最新值的Reactive操作符及实现方案

我明白你要实现的是类似「闸门」的逻辑：让数据源（Source0）的最新值先暂存起来，直到另一个开关流（Source1）发出允许信号（也就是示例里的T）时，再把这个最新值释放出来。结合你给出的Redux场景，我给你拆解下实现思路和代码：

核心思路拆解 #

你的需求本质是用开关流的「允许信号」触发获取数据源的最新值，具体分两步：

1. 先过滤开关流，只保留那些允许释放的信号（比如示例里的T）；
2. 每当收到允许信号时，立刻抓取数据源的最新值，推送到结果流里。

基础实现代码（匹配你的示例） #

假设你的数据源是SOME_TYPE的action流，开关流是store里的saving状态（当saving为true时对应示例里的T），代码可以这么写：

import { filter, withLatestFrom, map, distinctUntilChanged } from 'rxjs/operators';

// 1. 构建开关流：监听store中saving状态，只保留允许释放的信号（这里假设saving为true时允许）
const allowRelease$ = store.select(state => state.REDUCER.saving).pipe(
  filter(isAllowed => isAllowed), // 只保留true的允许信号
  distinctUntilChanged() // 避免连续重复的true触发多次无效操作
);

// 2. 数据源：你要缓存的SOME_TYPE action流
const sourceActions$ = action$.ofType('SOME_TYPE');

// 3. 构建结果流：每当开关允许时，推送数据源的最新值
const result$ = allowRelease$.pipe(
  withLatestFrom(sourceActions$), // 将开关信号与数据源最新值组合
  map(([_, latestAction]) => latestAction) // 只提取数据源的值，丢弃开关的布尔标记
);

对应你的示例流程验证： #

• Source0发出0→1→2→3，Source1前四个是F，第五个是T：此时开关触发，抓取Source0的最新值1，结果流发出1；
• 之后Source1回到F，直到下一个T：此时Source0最新值是2，结果流发出2；
• 接着Source1连续两个T：第一个T抓取最新值3，第二个T因为Source0没有新值，还是发出3，和你的示例结果完全匹配。

进阶场景：开关打开期间持续推送新值 #

如果你的需求是开关打开（T）期间，数据源的新值要实时推送，而不是只在开关触发时推一次，那可以用BehaviorSubject缓存最新值，再用开关控制是否转发：

import { BehaviorSubject, EMPTY } from 'rxjs';
import { switchMap, filter } from 'rxjs/operators';

// 用BehaviorSubject缓存数据源的最新值（初始值可根据业务设置默认值）
const cachedSource$ = new BehaviorSubject(null);
action$.ofType('SOME_TYPE').subscribe(cachedSource$);

// 开关流复用之前的定义
const allowRelease$ = store.select(state => state.REDUCER.saving).pipe(
  distinctUntilChanged()
);

// 结果流：开关打开时转发缓存流，关闭时静默
const result$ = allowRelease$.pipe(
  switchMap(isAllowed => {
    if (isAllowed) {
      return cachedSource$.pipe(filter(val => val !== null)); // 过滤初始的null值
    } else {
      return EMPTY; // 开关关闭时不发出任何值
    }
  })
);

这个版本中，只要开关处于T状态，数据源的新值会实时推送到结果流；开关切回F后，新值会被缓存，直到下次开关打开。

小提示 #

• 如果数据源可能还没发出过值，withLatestFrom不会触发结果流，这时候可以给数据源加startWith(默认值)来处理空流情况；
• 如果你用的是RxJS 6+，记得用pipeable操作符（就是上面代码里的.pipe()写法），不要用旧的链式操作符。

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