大文件异步迭代的生成器

Question

假设我有一个名为 openShapeFile 的函数，它读取一个文件，并生成一个包装源对象的 Promise，该对象具有一个 read 函数，它返回一个包装 Shapefile 中的实际值的 Promise，并具有.done 布尔值，可用于判断是否已到达文件末尾。

实际上，shapefile.open 来自这里： https://www.npmjs.com/package/shapefile

如果我现在想将文件读入数据库，我可以说：

openShapeFile(`shapefile.shp`).then((source) => source.read()
  .then(function log(result) {
    if (result.done) {
      return
    } else {
      let query = `INSERT INTO geodata(geometry, id, featcode) VALUES(ST_GeomFromGeoJSON('${
        JSON.stringify(Object.assign({}, result.value.geometry, {coordinates: result.value.geometry.coordinates.map(JSON.stringify)}))
      }'), '${
        result.value.properties.ID
      }', ${
        result.value.properties.FEATCODE
      });`
      query = query.split('"[[').join('[[').split(']]"').join(']]')
      return pool.query(query).then((result) => {
        return source.read().then(log)
      })
      }
      })).then(() => console.log(dirCount)).catch(err => 'Problem here')))

这只是工作，但有一个递归的承诺（奇怪）

因此，作为一个练习和/或看看它是否会产生更清晰的效果，我决定将其重写为生成器，产生如下内容：

function *insertQuery(query) {
    const result = pool.query(query)
    return result
  }

  const shapeFileGenerator = co.wrap(function* (source) {
    while (true) {
      const result = yield source.read()
      if (result.done) {
        return yield {}
      } else {
        let query = `INSERT INTO geodata(geometry, id, featcode) VALUES(ST_GeomFromGeoJSON('${
          JSON.stringify(Object.assign({}, result.value.geometry, {coordinates: result.value.geometry.coordinates.map(JSON.stringify)}))
        }'), '${
          result.value.properties.ID
        }', ${
          result.value.properties.FEATCODE
        });`
        query = query.split('"[[').join('[[').split(']]"').join(']]')
        yield* insertQuery(query)
      }
    }
  })
openShapeFile(`shapefile.shp`).then((source) => {
  const s = shapeFileGenerator(source)
})))

现在可以了！它读取所有数据！ 但是，我有点讨厌无限循环，而且我从不直接调用 .next。我怎样才能重做这个？用生成器做这样的事情的惯用方法是什么？似乎我应该能够使用s.next() 编写一个合适的生成器，从而生成source.read()？

Answer 1

您可以将您的逻辑编码为同步的，并通过顺序执行器nsynjs 执行。以下是在this file 上测试的稍作修改的工作示例：

main.js:

var nsynjs = require('nsynjs');
var shapefile = require('shapefile');

function synchrobousCode(shapefile /*, pool */) {
    var source = shapefile.open('UScounties.shp').data;
    var result = source.read().data;
    while(result && !result.done) {
        var query = "INSERT INTO geodata(geometry, id, featcode) VALUES('" +
            JSON.stringify(Object.assign({}, result.value.geometry, {coordinates: result.value.geometry.coordinates.map(JSON.stringify)})) +
            "'), '" +
            result.value.properties.ID +
            "'," +
            result.value.properties.FEATCODE +
            "')";
        console.log(query.length);
        // uncomment line below to sequentially insert to the DB
        // var queryRes = pool.query(query).data;
        result = source.read().data;
    }
}

nsynjs.run(synchrobousCode,null,shapefile /*, pool */ ,function () {
    console.log('all done');
})

Nsynjs 会自动检测某些函数调用是否返回 promise。如果是，它会等待promise解决，把它的结果放到data属性，然后才继续下一个表达式。

Answer 2

我会写

async function readFileToDB(filename) {
    const source = await openShapeFile(filename);
    for (let {value, done} = await source.read(); !done; {value, done} = await source.read()) {
        const query = `INSERT INTO geodata(geometry, id, featcode) VALUES(ST_GeomFromGeoJSON('${
            JSON.stringify(value.geometry)
        }'), '${
            value.properties.ID
        }', ${
            value.properties.FEATCODE
        });`
        const result = await pool.query(query);
    }
    console.log(dirCount);
}
readFileToDB(`shapefile.shp`).catch(err => console.error('Problem here', err));

我不认为递归解决方案有什么问题。

看来我应该能够使用s.next() 编写一个合适的生成器，从而生成source.read()？

不，发电机是同步的。不过，您可能想看看async iteration proposal。