0

试图了解如何为 Firebase 加载程序使用此异步流。我正在寻找：

  • 当我启用 Firebase Promise 时，即使最后一条记录输出到标准输出，该进程仍保持打开状态。
  • 使用这样注释的 Firebase 代码，该过程将关闭，但最终事件不会写入控制台。

所以我猜 Firebase 会留下一些需要在进程退出之前关闭的连接，如果我能弄清楚如何检测管道的末端，我可能会这样做。

我现在已经评论了 Firebase 的内容以将其从问题中删除，但是鉴于此代码，我如何在管道完成后运行一些代码？因为它是 done 功能没有执行。

const csv = require('csv');
const fs = require('fs');
const firebase = require('firebase-admin');
const config = require('../../app-config');
const serviceAccount = require('../../service-account');
const jobConf = require('../../data/category.jobfile');
const _ = require('lodash');

 // configure firebase
 firebase.initializeApp({
  credential: firebase.credential.cert(serviceAccount),
  databaseURL: config.firebase.databaseURL,
});
console.log('Loading config', jobConf);

const counts = {
  total: 0,
  new: 0,
  updated: 0,
}

const transformer = (record, done) => {
  const newObj = {};
  const fieldMaps = jobConf.fieldMaps;
  Object.keys(record).forEach((key) => {
    const destination = fieldMaps[key];
    const value = _.get(record, key);
    destination ? _.set(newObj, destination, value) : _.set(newObj, key, value);
  });
  Promise.resolve()
    .then(() => {
      done(null, JSON.stringify(newObj, null, 2));
    });
  // // now we look for existing row
  // firebase.database().ref(`/categories`).orderByChild('identifier').equalTo(newObj.identifier)
  //   .once('value')
  //   .then((snap) => snap.val())
  //   .then((val) => {
  //     if (val) {
  //       console.log('Got value', val);
  //       newObj.lastUpdated = firebase.database.ServerValue.TIMESTAMP;
  //     } else {
  //       newObj.created = firebase.database.ServerValue.TIMESTAMP;
  //       newObj.lastUpdated = firebase.database.ServerValue.TIMESTAMP;
  //       counts.total++;
  //       counts.new++;
  //       done(null, JSON.stringify(newObj, null, 2));
  //     }
  //   })
  //   .catch((err) => {
  //     console.log(err);
  //     done(err);
  //   })
};

const stream = fs.createReadStream(`../../data/${jobConf.sourceFile}`);
const pipe = stream.pipe(csv.parse({columns: true})).pipe(csv.transform(transformer)).pipe(process.stdout);

pipe.on('finish', function () { // not executed
  console.log('Done.');
  console.log('Stats: ', counts);
  process.exit(0);
});
4

0 回答 0