使用RxJS实现累积数组长度达N时停止的逻辑为何无输出？

原逻辑代码 #

原逻辑通过轮询生成器，筛选符合条件的值直到数组达到指定长度：

function* dataGeneratorFactory() {
    while (true) {
        for (const v of "abcdefghik") {
            yield v;
        }
    }
}

const shouldExclude = (letter) => letter === "b" || letter === "e";
const items = [];
const maxItemsLength = 10;
const dataGenerator = dataGeneratorFactory();

while (items.length < maxItemsLength) {
    const nextItem = dataGenerator.next();
    if (shouldExclude(nextItem.value)) continue;
    items.push(nextItem.value);
}

console.log(items);

目标与约束 #

将上述逻辑转换为RxJS模式，不可修改dataGeneratorFactory函数。

存在问题的代码 #

// maxItemsLength, dataGenerator, and dataGeneratorFactory are
// from the code snippet above
import { from, map, mergeAll, range, takeWhile } from "rxjs";

const observeable = range(0, maxItemsLength).pipe(
    map(() => from(dataGenerator).pipe(takeWhile(shouldExclude))),
    mergeAll()
);

observeable.subscribe({
    next: (e) => console.log(e),
});

问题原因 #

• takeWhile逻辑完全颠倒：takeWhile(shouldExclude)会在值满足排除条件时继续取值，一旦遇到不满足排除条件的值就停止订阅，但你需要的是排除满足条件的值、保留有效数据，逻辑完全搞反。
• 冗余的range+map+mergeAll组合：from(dataGenerator)本身就能持续从生成器取值，用range重复创建Observable毫无必要，还会导致生成器被多次订阅（生成器是迭代器，只能被遍历一次，后续订阅无法获取值）。
• 缺少有效数据数量控制：原逻辑是获取到maxItemsLength个有效数据就终止，当前代码没有实现这个核心终止条件。

修正后的代码 #

import { from, filter, take } from "rxjs";

const dataGenerator = dataGeneratorFactory();
const maxItemsLength = 10;
const shouldExclude = (letter) => letter === "b" || letter === "e";

from(dataGenerator)
  .pipe(
    filter(value => !shouldExclude(value)), // 过滤掉需要排除的值，保留有效数据
    take(maxItemsLength) // 获取到指定数量的有效数据后自动终止
  )
  .subscribe({
    next: (e) => console.log(e),
    complete: () => console.log('已获取全部目标数据')
  });

代码说明 #

• from(dataGenerator)：直接将生成器转换为Observable，持续从生成器中取值。
• filter(value => !shouldExclude(value))：反转排除逻辑，只保留不需要排除的有效数据。
• take(maxItemsLength)：当获取到指定数量的有效数据后自动终止Observable，完美匹配原逻辑的终止条件。

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