如何实现异步生成器以使用节点可读流进行流式传输？答案

【问题标题】：How to implement an async generator to stream with node Readable stream?如何实现异步生成器以使用节点可读流进行流式传输？
【发布时间】：2019-07-18 22:17:55
【问题描述】：

我想做这样的事情

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

function generatorToStream(generator) {
  return new Readable({
    read() {
      (async () => {
        for await (const result of generator()) {
          if (result.done) {
            this.push(null);
          } else {
            this.push(result.value);
          }
        }
      })();
    }
  });
}

generatorToStream(async function*() {
  const msg1 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg1;
  const msg2 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg2;

  const msg3 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg3;
}).pipe(process.stdout);

但它不起作用，end 事件从未被调用，我的终端上也没有收到任何数据。

关于如何实现它的任何解决方案或提示？

【问题讨论】：

您是否遇到任何错误，尤其是未处理的承诺拒绝？尝试将 .catch(console.error) 附加到您的 IIAFE。
IIRC 当您使用for await of 时，result 不会是迭代记录，而是值本身。生成器完成后，循环就结束了。
为每个 Promise 分别调用的 async function 和 await 就像一个生成器。甚至可以将它们结合起来吗？

标签： javascript node.js promise async-await

【解决方案1】：

我是Scramjet 的作者，这是一个功能性流处理框架，对您来说可能是一个简单的解决方案。

如果您可以在项目中添加总共 3 个依赖项，那再简单不过了：

const {StringStream} = require("scramjet");

StringStream.from(async function* () {
    yield await something();
    ...
});

如果您想自己实现它，请查看DataStream line 112 中的源代码 - 它应该很容易实现。一般来说，你需要实现这样的东西：

function generatorToStream(gen) {
    // obtain an iterator
    const iter = await gen();
    // create your output
    const out = new Passthrough();

    // this IIFE will do all the work
    (async () => {
        let done = false;
        for await (let chunk of iter) {
            // if write returns true, continue, otherwise wait until out is drained.
            if (!out.write(chunk)) await new Promise((res, rej) => this.once("drain", res);
        }
    })()
        // handle errors by pushing them to the stream for example
        .catch(e => out.emit('error', e));

    // return the output stream
    return out;
}

上面的例子或多或少是在超燃冲压发动机中发生的事情——那里有更多的优化来保持更少的事件处理程序等等，但上面的例子在一个简单的情况下应该可以很好地工作。

【讨论】：