Node JS 关闭一个读取流

Question

我坚持关闭读取流。我正在使用 csv-parser 模块从 CSV 文件中读取数据，进行一些处理并将数据写入 MongoDB。一切正常，除了我无法退出我的程序。它只是等待，我必须强制退出它。我怎样才能完成它的执行？const main = () => {  const records = [];  fs.readdir(dataPath, (err, files) => {    if (err) console.log("Failed to read. ", err);    else {      fs.createReadStream(`${dataPath}/${files[0]}`)        .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))        .on("data", data => records.push(data))        .on("end", async () => await saveToDB(getSysInfo(files[0]), records));    }  });};main();我尝试.on("close")在结束后添加一个事件，但这也无济于事。

Answer 1

这是我解决它的方法。问题在于底层 MongoDB 连接本身。流没有错。从文件中读取记录后，将它们插入数据库，并在读取并插入所有文件中的所有记录后，关闭底层连接以结束程序。

const getRecordsFromFile = fileName => {

  return new Promise((resolve, reject) => {

    const rows = [];

    fs.createReadStream(fileName)

      .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))

      .on("data", row => rows.push(row))

      .on("end", () => resolve(rows));

  });

};

const main = async () => {

  const files = fs.readdirSync(dataPath);

  for (let i = 0; i < files.length; i++) {

    const records = await getRecordsFromFile(`${dataPath}/${files[i]}`);

    await loadRecordsToDB(getSysInfo(files[i]), records);

  }

  mongoose.connection.close();

};

main();

Answer 2

对于 CLI，您必须使用 db.close() 关闭 mongodb 连接/您可以关闭。

句法：

// any other clean ups

    mongoose.connection.close(function () {

      console.log('Mongoose connection disconnected');

    });

重构代码：

const util = require("util");

const readDir = util.promisify(fs.readdir);

const readCSV = () => {

  return new Promise((res, rej) => {

    let records = [];

    fs.createReadStream(filePath)

      .pipe(

        csv({

          skipLines: 7,

          mapHeaders: ({ header, index }) => _.camelCase(header)

        })

      )

      .on("data", data => {

        records.push(data);

      })

      .on("error", error => {

        rej(data);

      })

      .on("end", () => {

        res(records);

      });

  });

};

const main = async () => {

  try {

    const files = await readDir(dataPath);

    await Promise.all(

      files.map(file => {

        return readCSV(`${dataPath}/${file}`) // read csv

          .then(csv => saveToDB(getSysInfo(file), csv)); // save in db

      })

    );

  } catch (error) {

    console.error(error);

  } finally {

    db.close(); // close mongo db

  }

};

main();

Answer 3

这是对您的答案的增强，它为 readStream 和两个await操作添加了错误处理，因此如果有任何错误，您的程序仍然可以以受控方式结束并正确关闭数据库：

const getRecordsFromFile = fileName => {

  return new Promise((resolve, reject) => {

    const rows = [];

    fs.createReadStream(fileName)

      .pipe(csv({ skipLines: 7, mapHeaders: ({ header, index }) => _.camelCase(header) }))

      .on("data", row => rows.push(row))

      .on("end", () => resolve(rows));

      .on("error", reject);                      // <==

  });

};

const main = async () => {

  const files = fs.readdirSync(dataPath);

  try {

    for (let i = 0; i < files.length; i++) {

      const records = await getRecordsFromFile(`${dataPath}/${files[i]}`);

      await loadRecordsToDB(getSysInfo(files[i]), records);

    }

  } catch(e) {                                             // <==

      console.log(e);                                      // <==

  } finally {                                              // <==

      // make sure we always close the connection

      mongoose.connection.close();

  }

};

main();