async await或Promise不返回流事件

Question

我有以下函数返回一个Promise,其中函数参数是一个异步函数:

  createObjectFrom(record) {
    const self = this;
    return new Promise(async (resolve, reject) => {
      let obj = {};

      for(let i = 0, l = self.opts.transformers.length; i < l; i++) {
        let transformer = self.opts.transformers[i];

        const headerIndex = findIndex(self.headers, (header) => {
          return header === transformer.column;
        });

        let csvValue = record[headerIndex];

        const lookUp = transformer.options.lookUp;

        const whereClause = {};

        whereClause[lookUp.column] = csvValue;

        console.log('before await');

        const result = await self.knex(lookUp.table).where(whereClause).select(lookUp.scalar);

        console.dir(result);

        obj[transformer.field] = result[0][lookUp.scalar];
      }

      return resolve(obj);
    });
  }

如果我在测试时调用这样的函数,它都会正确执行:

  it('creates the transformed object', async () => {
    const csvRecord = ['PREMIER', '07/11/1998', manager, 'Liverpool', 'Wimbledon', 0, 1, 'A', 0, 1, 'A'];

    const record = await transformer.createObjectFrom(csvRecord);

    expect(record.division).to.equal('PREMIER');
  }

但是在从csv-parse创建的流引发createObjectFrom的可读事件期间调用函数时:

  onReadable() {
    let record = this.parser.read();

    if (record === null) {
      return;
    }

    if (this.parser.count <= 1) {
      this.headers = record;
    } else {
      const recordPromises = this.createObjectFrom(record);
      this.recordPromises.push( newRecord );
    }
  }

代码进入console.log下面的声明createObjectFrom

  console.log('before here');
  const result = await self.knex(lookUp.table).where(whereClause).select(lookUp.scalar);
  console.dir(result);

但是console.dir由于Promise似乎没有解决,所以没有达到下面的声明.

如果我createObjectFrom从流处理之外的测试中调用,则它会正确解析.

我也试过重构异步等待退出一个承诺,但它仍然破产.

If I console.dir the promises on the [end][3] event of the stream they look like this:

[ Promise {
    _bitField: 0,
    _fulfillmentHandler0: undefined,
    _rejectionHandler0: undefined,
    _promise0: undefined,
    _receiver0: undefined },
  Promise {
    _bitField: 0,
    _fulfillmentHandler0: undefined,
    _rejectionHandler0: undefined,
    _promise0: undefined,
    _receiver0: undefined } ]

我有这个包含源代码和失败测试的repo.

我很困惑发生了什么事.

以下测试也通过,所以它肯定与流有关:

  it('creates the transformed object', async () => {
    const csvRecords = [
      ['PREMIER', '07/11/1998', manager, 'Liverpool', 'Wimbledon', 0, 1, 'A', 0, 1, 'A'],
      ['PREMIER', '11/11/1998', manager, 'QPR', 'Sunderland',3,3, 'Sunderland',0,0,'Sunderland'],
      ['PREMIER', '14/11/1998', manager, 'Southampton', 'Liverpool', 3, 3, 'D', 0, 0, 'D']
    ];

    for(var i = 0, l = csvRecords.length; i < l; i++) {
      const csvRecord = csvRecords[i];

      const record = await transformer.createObjectFrom(csvRecord);

      expect(record.division).to.equal('PREMIER');

      expect(record.manager_id).to.equal(manager_id);
    }
  }

Answer 1

为什么你不这样改变你的代码：

createObjectFrom: async (record) => {
    const self = this;
    // return new Promise(async (resolve, reject) => {
      let obj = {};

      for(let i = 0, l = self.opts.transformers.length; i < l; i++) {
        let transformer = self.opts.transformers[i];

        const headerIndex = findIndex(self.headers, (header) => {
          return header === transformer.column;
        });

        let csvValue = record[headerIndex];

        const lookUp = transformer.options.lookUp;

        const whereClause = {};

        whereClause[lookUp.column] = csvValue;

        console.log('before await');

        const result = await self.knex(lookUp.table).where(whereClause).select(lookUp.scalar);

        console.dir(result);

        obj[transformer.field] = result[0][lookUp.scalar];
      }
      return obj;
      // return resolve(obj);
    // });
  }