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

这是对您答案的增强，它为 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();

Answer 2

这是我解决它的方法。问题在于底层 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 3

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

语法：

// 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();