node.js - 如何使用事件发射器作为异步生成器

Question

我正在尝试使用 babel 的异步生成器的简洁语法(我被 Node 8 困住了)，我想知道如何将事件发射器干净地转换为异步生成器

到目前为止我得到的看起来像这样

    const { EventEmitter } = require('events')

    // defer fonction for resolving promises out of scope
    const Defer = () => {
      let resolve
      let reject
      let promise = new Promise((a, b) => {
        resolve = a
        reject = b
      })
      return {
        promise,
        reject,
        resolve
      }
    }


    // my iterator function
    function readEvents(emitter, channel) {
      const buffer = [Defer()]
      let subId = 0

      emitter.on(channel, x => {
        const promise = buffer[subId]
        subId++
        buffer.push(Defer())
        promise.resolve(x)
      })

      const gen = async function*() {
        while (true) {
          const val = await buffer[0].promise
          buffer.shift()
          subId--
          yield val
        }
      }

      return gen()
    }

    async function main () {
      const emitter = new EventEmitter()
      const iterator = readEvents(emitter, 'data')

      // this part generates events
      let i = 0
      setInterval(() => {
        emitter.emit('data', i++)
      }, 1000)

      // this part reads events
      for await (let val of iterator) {
        console.log(val)
      }
    }

    main()

这很不方便 - 可以简化吗？

Answer 1

我想出了这个:

async *stream<TRecord extends object=Record<string,any>>(query: SqlFrag): AsyncGenerator<TRecord> {
    const sql = query.toSqlString();

    let results: TRecord[] = [];
    let resolve: () => void;
    let promise = new Promise(r => resolve = r);
    let done = false;

    this.pool.query(sql)
        .on('error', err => {
            throw err;
        })
        .on('result', row => {
            results.push(row);
            resolve();
            promise = new Promise(r => resolve = r);
        })
        .on('end', () => {
            done = true;
        })

    while(!done) {
        await promise;
        yield* results;
        results = [];
    }
}

到目前为止似乎正在工作。

即您像 Khanh 的解决方案一样创建了一个虚拟 promise ，以便您可以等待第一个结果，但随后由于许多结果可能会同时出现，因此您将它们插入一个数组并重置 promise 以等待结果(或一批)结果)。即使这个 promise 在等待之前被重写数十次也没关系。

然后我们可以使用 yield* 一次生成所有结果，并刷新下一批的数组。