Node.js 流可以作为协程吗？答案

【问题标题】：Can Node.js stream be made as coroutine?Node.js 流可以作为协程吗？
【发布时间】：2016-09-10 14:06:49
【问题描述】：

有没有办法让 Node.js 流作为协程。

示例斐波那契数字流。

fibonacci.on('data', cb);
//The callback (cb) is like
function cb(data)
{
    //something done with data here ...
}

期待

function* fibonacciGenerator()
{
    fibonacci.on('data', cb);
    //Don't know what has to be done further... 
};

var fibGen = fibonacciGenerator();
fibGen.next().value(cb);
fibGen.next().value(cb);
fibGen.next().value(cb);
.
.
.

从生成器中获取所需的数字。这里的斐波那契数列只是一个例子，实际上流可以是任何文件、mongodb 查询结果等。

也许是这样的

将“stream.on”函数设为生成器。
将 yield 放在回调函数中。
获取生成器对象。
调用 next 并获取流中的下一个值。

如果是的话，至少有可能吗？如果不是，为什么？也许是一个愚蠢的问题:)

【问题讨论】：

嗯，不。你不能这样做。您必须等到 async/await 标准化。
在python中是可能的...def sgen(filename): f = open(filename,'r') for l in f: yield f.readline() g = sgen('/home/datadumpfile.xml') print(next(g)) print(next(g)) print(next(g))

标签： node.js generator coroutine

【解决方案1】：

如果您不想使用转译器（例如 Babel）或等到 async/await 进入 Node.js，您可以使用生成器和 Promise 自己实现它。

缺点是您的代码必须存在于生成器中。

首先，您可以创建一个辅助函数，它接收流并返回一个函数，该函数在调用时返回流的下一个“事件”的承诺（例如data）。

function streamToPromises(stream) {
  return function() {
    if (stream.isPaused()) {
      stream.resume();
    }

    return new Promise(function(resolve) {
      stream.once('data', function() {
        resolve.apply(stream, arguments);
        stream.pause();
      });
    });
  }
}

当你不使用它时它会暂停流，并在你询问下一个值时恢复它。

接下来，你有一个助手，它接收一个生成器作为参数，每次它产生一个承诺时，它都会解析它并将其结果传递回生成器。

function run(fn) {
  var gen = fn();
  var promise = gen.next().value;

  var tick = function() {
    promise.then(function() {
      promise = gen.next.apply(gen, arguments).value;
    }).catch(function(err) {
      // TODO: Handle error.
    }).then(function() {
      tick();
    });
  }

  tick();
}

最后，您将在生成器中执行自己的逻辑，并使用 run 帮助器运行它，如下所示：

run(function*() {
  var nextFib = streamToPromises(fibonacci);

  var n;

  n = yield nextFib();
  console.log(n);

  n = yield nextFib();
  console.log(n);
});

您自己的生成器将产生 Promise，暂停其执行并将控制权传递给 run 函数。
run 函数将解析承诺并将其值传递回您自己的生成器。

这就是它的要点。您还需要修改 streamToPromises 以检查其他事件（例如 end 或 error）。

【讨论】：

【解决方案2】：

class FibonacciGeneratorReader extends Readable {
    _isDone = false;
    _fibCount = null;
    _gen = function *() {
        let prev = 0, curr = 1, count = 1;
        while (this._fibCount === -1 || count++ < this._fibCount) {
            yield curr;
            [prev, curr] = [curr, prev + curr];
        }
        return curr;
    }.bind(this)();

    constructor(fibCount) {
        super({
            objectMode: true,
            read: size => {
                if (this._isDone) {
                    this.push(null);
                } else {
                    let fib = this._gen.next();
                    this._isDone = fib.done;
                    this.push(fib.value.toString() + '\n');
                }
            }
        });

        this._fibCount = fibCount || -1;
    }
}

new FibonacciGeneratorReader(10).pipe(process.stdout);

输出应该是：

【讨论】：