node.js - 如何在以多部分形式处理文件流之前等待字段

Question

我正在使用 SendGrid 通过电子邮件接收文件。 SendGrid 解析传入的电子邮件并将文件以多部分形式发送到我设置的端点。

我不想要我本地磁盘上的文件，所以我将它们直接流式传输到 Amazon S3。这很完美。

但在流式传输到 S3 之前，我需要掌握目标邮件地址，以便计算出正确的 s3 文件夹。这是在表单帖子中名为“to”的字段中发送的。不幸的是，这个字段有时会在文件到达后到达，因此我需要一种方法来等待目标字段，然后才能准备好接收流。

我想我可以将 onField 包装在一个 promise 中，并等待 onFile 中的 to-field。但是当字段在文件之后到达时，这个概念似乎会自行锁定。

我是 booth 流和 promises 的新手。如果有人能告诉我如何做到这一点，我将不胜感激。

这是无效的伪代码:

function sendGridUpload(req, res, next) { 
  var busboy = new Busboy({ headers: req.headers });

  var awaitEmailAddress = new Promise(function(resolve, reject) {
    busboy.on('field', function(fieldname, val, fieldnameTruncated, valTruncated) {
      if(fieldname === 'to') {
        resolve(val);
      } else {
        return;
      }
    });
  });


  busboy.on('file', function(fieldname, file, filename, encoding, mimetype) {

    function findInbox(emailAddress) {
      console.log('Got email address: ' + emailAddress);

      ..find the inbox and generate an s3Key
      return s3Key;
    }

    function saveFileStream(s3Key) {
      ..pipe the file directly to S3
    }

    awaitEmailAddress.then(findInbox)
    .then(saveFileStream)
    .catch(function(err) {
      log.error(err)
    });
  });

  req.pipe(busboy);
}

Answer 1

我终于搞定了。解决方案不是很漂亮，我实际上已经切换到另一个概念(在帖子末尾描述)。

为了在“to”字段到达之前缓冲传入数据，我使用了@samcday 的stream-buffers。当我获得目标字段时，我将可读流释放到为数据排列的管道。

这是代码(省略了一些部分，但有必要的部分)。

var streamBuffers = require('stream-buffers');

function postInboundMail(req, res, next) {
  var busboy = new Busboy({ headers: req.headers});

  //Sometimes the fields arrives after the files are streamed.
  //We need the "to"-field before we are ready for the files
  //Therefore the onField is wrapped in a promise which gets
  //resolved when the to field arrives
  var awaitEmailAddress = new Promise(function(resolve, reject) {
    busboy.on('field', function(fieldname, val, fieldnameTruncated, valTruncated) {
      var emailAddress;

      if(fieldname === 'to') {
        try {
          emailAddress = emailRegexp.exec(val)[1]
          resolve(emailAddress)
        } catch(err) {
          return reject(err);        
        }
      } else {
        return;
      }
    });
  });


  busboy.on('file', function(fieldname, file, filename, encoding, mimetype) {
    var inbox;

    //I'm using readableStreamBuffer to accumulate the data before
    //I get the email field so I can send the stream through to S3
    var readBuf = new streamBuffers.ReadableStreamBuffer();

    //I have to pause readBuf immediately. Otherwise stream-buffers starts
    //sending as soon as I put data in in with put().
    readBuf.pause();

    function getInbox(emailAddress) {
      return model.inbox.findOne({email: emailAddress})
      .then(function(result) {
        if(!result) return Promise.reject(new Error(`Inbox not found for ${emailAddress}`))

        inbox = result;
        return Promise.resolve();
      });
    }

    function saveFileStream() {
      console.log('=========== starting stream to S3 ========= ' + filename)

      //Have to resume readBuf since we paused it before
      readBuf.resume();

      //file.save will approximately do the following:
      // readBuf.pipe(gzip).pipe(encrypt).pipe(S3)
      return model.file.save({
        inbox: inbox,
        fileStream: readBuf
      });
    }

    awaitEmailAddress.then(getInbox)
    .then(saveFileStream)
    .catch(function(err) {
      log.error(err)
    });


    file.on('data', function(data) {
      //Fill readBuf with data as it arrives
      readBuf.put(data);
    });

    file.on('end', function() {
      //This was the only way I found to get the S3 streaming finished.
      //Destroysoon will let the pipes finish the reading bot no more writes are allowed
      readBuf.destroySoon()
    });
  });


  busboy.on('finish', function() {
    res.writeHead(202, { Connection: 'close', Location: '/' });
    res.end();
  });

  req.pipe(busboy);
}

我真的很想收到有关此解决方案的反馈，即使我没有使用它。我觉得这可以做得更加简单和优雅。

新解决方案:我没有等待目标字段，而是直接将流发送到 S3。我想，我在传入流和 S3 保存之间放入的东西越多，由于我的代码中的错误而丢失传入文件的风险就越高。 (如果我没有以 200 响应，SendGrid 最终会重新发送文件，但这需要一些时间。)

我是这样做的:

1. 在数据库中为文件保存一个占位符
2. 将流通过管道传输到 S3
3. 在信息到达时用更多信息更新占位符

此解决方案还让我有机会轻松掌握不成功的上传，因为不成功上传的占位符将不完整。

//迈克尔