如何从另一个Readable Stream返回过滤后的流？除Transform外还有其他方法吗？

你的示例代码并没有实现预期的过滤功能——直接返回输入流后，监听data事件的回调返回值不会对流的输出产生任何影响，因为data事件仅用于通知数据到达，回调的返回值会被忽略。

除了使用stream.Transform，还有以下两种原生实现方式：

1. 手动封装可读流 #

创建一个新的Readable实例，手动读取输入流的内容，过滤后将符合条件的数据推送到新流中：

const { Readable } = require('stream');

const filterStream = (inputStream, regexp) => {
  return new Readable({
    async read() {
      let chunk;
      try {
        while ((chunk = await inputStream.read()) !== null) {
          if (regexp.test(chunk)) {
            // 推送符合条件的数据到新流
            if (!this.push(chunk)) {
              // 若流被暂停，等待drain事件恢复后继续
              await new Promise(resolve => this.once('drain', resolve));
            }
          }
        }
        // 输入流读取完毕，结束新流
        this.push(null);
      } catch (err) {
        this.emit('error', err);
      }
    }
  });
};

// 使用示例
const inputStream = Readable.from(["aaa", "aAa", "aab"]).setEncoding("utf8");
const outputStream = filterStream(inputStream, /aaa/i);

outputStream.on('data', data => console.log('过滤后的数据:', data));

2. 异步迭代器结合可读流转换 #

通过异步迭代器遍历输入流的所有数据，过滤后将结果转为新的可读流返回：

const { Readable } = require('stream');

const filterStream = async (inputStream, regexp) => {
  const filtered = [];
  // 遍历输入流的每个数据块
  for await (const chunk of inputStream) {
    if (regexp.test(chunk)) {
      filtered.push(chunk);
    }
  }
  // 将过滤后的数组转为可读流
  return Readable.from(filtered).setEncoding('utf8');
};

// 使用示例
(async () => {
  const inputStream = Readable.from(["aaa", "aAa", "aab"]).setEncoding("utf8");
  const outputStream = await filterStream(inputStream, /aaa/i);
  
  outputStream.on('data', data => console.log('过滤后的数据:', data));
})();

注意：第二种方法会将所有符合条件的数据先缓存到内存中，再转为流，适合数据量不大的场景；如果处理大文件等海量数据，第一种方法的流式处理更节省内存。

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