javascript - 异步等待或 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');
  }

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

  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 );
    }
  }

代码在 createObjectFrom

中获取下面的 console.log 语句

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

但没有到达下面的 console.dir 语句，因为 Promise 似乎没有解决。

如果我从流处理之外的测试中调用 createObjectFrom，那么它会正确解析。

我也尝试重构 async await 以返回一个 promise，但它仍然失败了。

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 } ]

我有this包含源代码和失败测试的 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);
    // });
  }