node.js - エラーの発生時にノード ストリームがハングする

Question

CSV をチェックするストリームがあります。応答を送り返した後でもエラーが発生した場合を除いて、正常に動作します。

export function ValidateCSV(options) {
  let opt = options;
  if (!(this instanceof ValidateCSV)) return new ValidateCSV(opt);

  if (!opt) opt = {};
  opt.objectMode = true;
  opt.highWaterMark = 1000000;
  Transform.call(this, opt);
}

util.inherits(ValidateCSV, Transform);

ValidateCSV.prototype.destroy = function () {
  this.readable = false;
  this.writable = false;
  this.emit('end');
};

ValidateCSV.prototype._transform = function (chunk, encoding, done) {
  // Do some stuff to the chunk
  // Emit error
    if (required.length > 0) {
        this.emit('error', `The following columns are required: ${required.join(', ')}`);
    }

  done();
};

destroy メソッドを追加することで修正できましたが、それでも遅く、数秒間ハングします。変換ストリームを終了/破棄するより良い方法はありますか?

ValidateCSV.prototype.destroy = function () {
    this.readable = false;
    this.writable = false;
    this.emit('end');
  };

編集：

バスボーイでストリームを使用する方法は次のとおりです。

function processMultipart(req, res) {
  const userId = req.query._userId;
  const busboy = new Busboy({ headers: req.headers, limits: { files: 1 } });
  const updateId = req.params.id;

  // Transform stream to validate the csv
  const validateCSV = new ValidateCSV();
  validateCSV
    .on('finish', () => {
      // Process the csv
    })
    .on('error', (er) => {
      //Do some logging
      res.status(500).json(er).end();
    });

    // Multipart upload handler
    busboy
      .on('file', (fieldname, file, filename) => {
        dataset.name = fieldname.length > 0 ?
          fieldname : filename.substr(0, filename.indexOf('.csv'));
        file
          .on('error', (er) => {
            //Send Error
          })
          .on('end', () => {
            // Save dataset to mongo
            if (dataset._update) {
              res.status(200).json(dataset).end();
            } else {
              Dataset.create(dataset, (er) => {                
                if (er) {
                  res.status(500).json(er).end();
                } else {
                  res.status(200).json(dataset).end();
                }
              });
            }
          }).pipe(validateCSV);
      });
    req.pipe(busboy);
}