如何根据另一observable的布尔值控制指定observable的项推送

你需要的是基于另一Observable状态的阀门过滤逻辑，Rx原生操作符即可实现，不需要自定义复杂逻辑，以下是两种主流实现方式：

1. 原生操作符组合方案（全Rx平台兼容） #

用withLatestFrom + filter + map的组合，是兼容性最好的实现，所有Rx语言版本（RxJS、RxJava、Rx.NET等）都支持：
核心逻辑：

• 每次Observable a 发射项时，取Observable b 最近发射的布尔值和当前a的项组合为二元组
• 过滤出二元组中b值为true的组合
• 提取a的原始值传递到下游

RxJS示例代码：

import { withLatestFrom, filter, map, startWith } from 'rxjs';

// 如果需要默认关闭阀门，给b添加startWith(false)，默认打开就传true
const controlledB = b.pipe(startWith(false));

const filteredA = a.pipe(
  withLatestFrom(controlledB),
  filter(([aItem, bState]) => bState),
  map(([aItem]) => aItem)
);

RxJava示例代码逻辑完全一致：

Observable<String> filteredA = a
    .withLatestFrom(b.startWith(false), (aItem, bState) -> new Pair<>(aItem, bState))
    .filter(pair -> pair.second)
    .map(pair -> pair.first);

注意：如果不给b设置初始值，在b发射第一个值之前，a的所有项都会被拦截，不会向下游传递。

2. 封装操作符方案（适合高频使用场景） #

大部分Rx扩展库已经把上述逻辑封装为pausable（可暂停）操作符，直接传入控制用的Observable b 即可，代码更简洁：
RxJS扩展库示例：

import { pausable } from 'rxjs-pausable';

const filteredA = a.pipe(
  pausable(b)
);

常见边界场景适配 #

• 未收到b的第一个值时默认允许a的项通行：将startWith(false)改为startWith(true)
• b切换为true时需要补发最近一次被拦截的a的项：给a添加replay(1)操作符缓存最近值
• 多订阅场景下避免a重复生产数据：给a添加share()操作符共享数据源

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