使用 Web Streams，从多个 TransformStreams 创建一个 TransformStream

Question

是否可以使用 whatwg 流（网络 Streams API）从其他几个 TransformStreams 中创建一个 TransformStream？

例如，如果我有两个按顺序运行的 TransformStreams，如 transformer1 和 transformer2：

readable.pipeThrough(transformer1).pipeThrough(transformer2).pipeTo(writable)

最终，我希望能够将其转换为

readable.pipeThrough(allTransformers).pipeTo(writable)

其中 allTransformers 是组合 transformer1 和 transformer2 的 TransformStream。

下面不是真正的功能代码，但我认为会有一种方法可以做这样的事情：

const allTransformers = transformer1.pipeThrough(transformer2)

这显然是一个简化的例子，但你可以想象有许多转换流，我想重构为一个单一的、可重用的转换管道。

Answer 1

我有同样的问题，这是我的解决方案。

在下面的示例中，UpperCaseReadableStream 通过TextDecoderStream、UpperCaseTextStream 和TextEncoderStream 传输流。

class UpperCaseTextStream extends TransformStream {
  constructor() {
    super({
      transform(chunk, controller) {
        controller.enqueue(chunk.toUpperCase());
      }
    });
  }
}

class UpperCaseReadableStream {
  constructor(...strategies) {
    const { writable, readable } = new TransformStream({}, ...strategies);
    this.writable = writable;
    this.readable = readable
      .pipeThrough(new TextDecoderStream())
      .pipeThrough(new UpperCaseTextStream())
      .pipeThrough(new TextEncoderStream());
  }
}

const TEXT_CHUNK =
 "Lorem ipsum dolor sit amet, consectetuer adipiscing elit, sed diam nonummy " +
 "nibh euismod tincidunt ut laoreet dolore magna aliquam erat volutpat.";
const TEXT_CONTENT = new Array(1024 * 1024).fill(TEXT_CHUNK).join("");
const readableInput = new Blob([TEXT_CONTENT]).stream();
const readableOuput = readableInput.pipeThrough(new UpperCaseReadableStream());  
(async () => {
  const text = await new Response(readableOuput).text();
  console.log("Test OK?", text == TEXT_CONTENT.toUpperCase());
})().catch(console.error);

这是使用类 PipelineStream 的更通用的实现，它扩展 TransformStream 并接受 TransformStream 实例数组作为第一个参数。

class UpperCaseTextStream extends TransformStream {
  constructor() {
    super({
      transform(chunk, controller) {
        controller.enqueue(chunk.toUpperCase());
      }
    });
  }
}

class PipelineStream extends TransformStream {
  constructor(transformStreams, ...strategies) {
    super({}, ...strategies);

    const readable = [super.readable, ...transformStreams]
      .reduce((readable, transform) => readable.pipeThrough(transform));

    Object.defineProperty(this, "readable", {
      get() {
        return readable;
      }
    });
  }
}

const TEXT_CHUNK =
 "Lorem ipsum dolor sit amet, consectetuer adipiscing elit, sed diam nonummy " +
 "nibh euismod tincidunt ut laoreet dolore magna aliquam erat volutpat.";
const TEXT_CONTENT = new Array(1024 * 1024).fill(TEXT_CHUNK).join("");
const readableInput = new Blob([TEXT_CONTENT]).stream();
const readableOuput = readableInput.pipeThrough(new PipelineStream([
  new TextDecoderStream(),
  new UpperCaseTextStream(),
  new TextEncoderStream()]));
(async () => {
  const text = await new Response(readableOuput).text();
  console.log("Test OK?", text == TEXT_CONTENT.toUpperCase());
})().catch(console.error);