Meteor：如何将大文件流式传输并解析到异步节点函数？

Question

我正在使用job-collection 包执行以下操作：

1. 下载包含大量网页元数据的大文件
2. 使用 NPM event-stream 包从由正则表达式拆分的文件元数据创建流
3. 检查集合中的元数据是否匹配（我一直在尝试将每个网页的元数据流式传输到另一个函数来执行此操作）

文件太大，无法缓冲，因此需要流式传输。 Here is a small file with a few examples of the metadata 如果你想试试这个。

job-collection 包中的每个作业都已经在一个异步函数中：

var request = Npm.require('request');
var zlib = Npm.require('zlib');
var EventStream = Meteor.npmRequire('event-stream');

function (job, callback) {

//This download is much too long to block
  request({url: job.fileURL, encoding: null}, function (error, response, body) {
    if (error) console.error('Error downloading File');
    if (response.statusCode !== 200) console.error(downloadResponse.statusCode, 'Status not 200');

    var responseEncoding = response.headers['content-type'];
    console.log('response encoding is %s', responseEncoding);
    if (responseEncoding === 'application/octet-stream' || 'binary/octet-stream') {
      console.log('Received binary/octet-stream');
      var regexSplit = /WARC\/1\./;
      response.pipe(zlib.createGunzip()
              .pipe(EventStream.split(regexSplit))
              .pipe(EventStream.map(function (webpageMetaData) {
      /* Need parse the metaData or pass each webpageMetaData to function
       * This next function could block if it had to */
      searchPageMetaData(webpageMetaData); // pass each metadatum to this function to update a collection - this function can be synchronous
    }));
    } else {
      console.error('Wrong encoding');
    }
  });
}

function searchWebPageMetaData(metaData) {
  //  Parse JSON and search collection for match
}

• 有没有更好的方法来构建它？我在正确的轨道上吗？
• Meteor.bindEnvironment 应该放在哪里？ - 每次传递到searchWebPageMetaData() 时我是否绑定环境？我需要在这里明确使用纤维吗？
• 如果我将它运行到process.stdout，则流在运行时会停止。我应该将流放入 Meteor 的包装中吗？
• 我知道Meteor.wrapAsync。我想将最里面的searchWebPageMetaData() 函数包装在Meteor.wrapAsync 中吗？ （认为​​我在输入时会回答“是”）
• 流是否会变慢以补偿 DB 调用的缓慢？我的猜测是否定的，但我该如何处理？

我花了很长时间学习 Meteor 的 wrapAsync 和 bindEnvironment，但无法将它们整合在一起并理解在哪里使用它们。

补充 1

澄清一下，步骤如下：

1. 下载文件；
2. 创建流；
3. 解压；
4. 将其拆分为单独的网页 - EventStream 处理此问题
5. 将其发送到函数 - 不需要返回值；这可能是阻塞的，它只是一些搜索和数据库调用

我试图做这样的事情，除了我需要帮助的核心代码位于不同文件的函数中。以下代码包含@electric-jesus 的大部分答案。

   processJobs('parseWatFile', {
     concurrency: 1,
     cargo: 1,
     pollInterval: 1000,
     prefetch: 1
   }, function (job, callback) {

     if (job.data.watZipFileLink) {
       queue.pause();
       console.log('queue should be paused now');


       var watFileUrl = 'https://s3.amazonaws.com/ja-common-crawl/exampleWatFile.wat.gz';
       function searchPageMetaData(webpageMetaData, callback) {
         console.log(webpageMetaData);  // Would be nice to just get this function logging each webPageMetaData
         future.return(callback(webpageMetaData));  //I don't need this to return any value - do I have to return something?
     }

      if (!watFile)
        console.error('No watFile passed to downloadAndSearchWatFileForEntity ');

      var future = new Future(); // Doc Brown would be proud.

      if(typeof callback !== 'function') future.throw('callbacks are supposed to be functions.');

    request({url: watFile, encoding: null}, function (error, response, body) {

      if (error)                        future.throw('Error Downloading File');
      if (response.statusCode !== 200)  future.throw('Expected status 200, got ' + response.statusCode + '.');

      var responseEncoding = response.headers['content-type'];

    if (responseEncoding === 'application/octet-stream' || 'binary/octet-stream') {

      var regexSplit = /WARC\/1\./;
      response.pipe(zlib.createGunzip()
        .pipe(EventStream.split(regexSplit))
        .pipe(EventStream.map(function (webpageMetaData) {
        searchPageMetaData(webpageMetaData, callback);
      })
    ));
    } else {
      future.throw('Wrong encoding');
    }
    });

    return future.wait();

    } else {
      console.log('No watZipFileLink for this job');
      job.log('ERROR: NO watZipFileLink from commonCrawlJob collection');
    }
      queue.resume();
      job.done;
      callback();
  }

Answer 1

有趣，看起来不错。我从未使用过job-collection，但它似乎只是一个 Mongo 驱动的任务队列.. 所以我假设它像一个常规队列一样工作。我一直在寻找带有回调的东西，我肯定会使用Future 模式。例如：

var request = Npm.require('request');
var zlib = Npm.require('zlib');
var EventStream = Meteor.npmRequire('event-stream');

var Future = Npm.require('fibers/future');


var searchWebPageMetaData = function (metaData) {
  //  Parse JSON and search collection for match
  // make it return something
  var result = /droids/ig.test(metaData);
  return result;
}

var processJob = function (job, callback) {

  var future = new Future(); // Doc Brown would be proud.

  if(typeof callback !== 'function') future.throw("Oops, you forgot that callbacks are supposed to be functions.. not undefined or whatever.");

  //This download is much too long to block
  request({url: job.fileURL, encoding: null}, function (error, response, body) {

    if (error)                        future.throw("Error Downloading File");
    if (response.statusCode !== 200)  future.throw("Expected status 200, got " + downloadResponse.statusCode + ".");

    var responseEncoding = response.headers['content-type'];


    if (responseEncoding === 'application/octet-stream' || 'binary/octet-stream') {

      var regexSplit = /WARC\/1\./;
      response.pipe(zlib.createGunzip()
              .pipe(EventStream.split(regexSplit))
              .pipe(EventStream.map(function (webpageMetaData) {
      /* Need parse the metaData or pass each webpageMetaData to function
       * This next function could block if it had to */

      // pass each metadatum to this function to update a collection - this function can be synchronous

      future.return(callback(webpageMetaData)); // this way, processJob returns whatever we find in the completed webpage, via callback.

    }));
    } else {
      future.throw('Wrong encoding');
    }
  });

  return future.wait();
}

示例用法：

所以每当你在这里分配变量时：

var currentJob = processJob(myjob, searchWebPageMetaData);

即使使用同步类型获取/变量分配，您也可以为您及时完成和传输异步工作。

回答你的问题，

• Meteor.bindEnvironment 应该放在哪里？ - 我每次传递给 searchWebPageMetaData() 时是否绑定环境？我需要在这里明确使用纤维吗？

  不是真的，我相信fibers/future 的显式使用已经解决了这个问题。

• 如果我将它运行到 process.stdout，则流在运行时会停止。我是否应该将流放入 Meteor 的包装之一中

  你是什么意思？我隐约记得 process.stdout 是阻塞的，这可能是一个原因。同样，将结果包装在 future 中应该可以解决这个问题。

• 我知道 Meteor.wrapAsync。我想在 Meteor.wrapAsync 中包装最里面的 searchWebPageMetaData() 函数吗？ （认为​​我在输入时会回答“是”）

  Take a look at the Meteor.wrapAsync helper code。它基本上是应用future 分辨率，当然你可以这样做，然后你也可以明确地单独使用fibers/future 没有问题。

• 流会变慢以补偿数据库调用的缓慢吗？我的猜测是没有，但我该如何处理呢？

  不太清楚你在这里的意思.. 但由于我们正在尝试使用异步光纤，我的猜测也不是。我还没有看到使用纤维的任何缓慢。可能仅当一次启动（并同时运行）多个作业时，您才会在内存使用方面遇到性能问题。保持并发队列较低，因为 Fibers 在同时运行东西方面可能非常强大。你只有一个核心来处理这一切，这是一个可悲的事实，因为节点不能多核:(

Answer 2

如果您想正确处理所有错误，这将非常棘手。所以一个人应该问自己，如果：你的代码抛出一个异常，或者error 事件处理程序被调用，该怎么办。您希望错误正确传播，即在光纤调用流代码中作为异常抛出。我为我们的job-collecton 任务之一实现了类似的东西，用于提取tar files。

首先你需要一些辅助函数：

bindWithFuture = (futures, mainFuture, fun, self) ->
  wrapped = (args...) ->
    future = new Future()

    if mainFuture
      future.resolve (error, value) ->
        # To resolve mainFuture early when an exception occurs
        mainFuture.throw error if error and not mainFuture.isResolved()
        # We ignore the value

    args.push future.resolver()
    try
      futures.list.push future
      fun.apply (self or @), args
    catch error
      future.throw error

    # This waiting does not really do much because we are
    # probably in a new fiber created by Meteor.bindEnvironment,
    # but we can still try to wait
    Future.wait future

  Meteor.bindEnvironment wrapped, null, self

wait = (futures) ->
  while futures.list.length
    Future.wait futures.list
    # Some elements could be added in meantime to futures,
    # so let's remove resolved ones and retry
    futures.list = _.reject futures.list, (f) ->
      if f.isResolved()
        # We get to throw an exception if there was an exception.
        # This should not really be needed because exception should
        # be already thrown through mainFuture and we should not even
        # get here, but let's check for every case.
        f.get()
        true # And to remove resolved

然后你可以运行类似的东西：

mainFuture = new Future()

# To be able to override list with a new value in wait we wrap it in an object
futures =
  list: []

bindWithOnException = (f) =>
  Meteor.bindEnvironment f, (error) =>
    mainFuture.throw error unless mainFuture.isResolved()

onWebpageMetaData = (metaData, callback) =>
  return callback null if mainFuture.isResolved()

  # Do whatever you want here.
  # Call callback(null) when you finish.
  # Call callback(error) if there is an error.
  # If you want to call into a Meteor code inside some other callback for async code you use,
  # use bindWithOnException to wrap a function and stay inside a Meteor environment and fiber.

  MeteorCollection.insert
    metaData: metaData

  callback null

requestFuture = new Future()
request
  url: job.fileURL
  encoding: null
, 
  (error, response, body) ->
    return requestFuture.throw error if error
    return requestFuture.throw new Error "Expected status 200, got #{ response.statusCode }." unless response.statusCode is 200
    requestFuture.return response

response = requestFuture.wait()

responseEncoding = response.headers['content-type']
throw new Error "Wrong encoding" unless responseEncoding in ['application/octet-stream', 'binary/octet-stream']

regexSplit = /WARC\/1\./

response.pipe(
  zlib.createGunzip()
).pipe(
  EventStream.split regexSplit
).pipe(
  EventStream.map bindWithFuture futures, mainFuture, onWebpageMetaData
).on('end', =>
  # It could already be resolved by an exception from bindWithFuture or bindWithOnException
  mainFuture.return() unless mainFuture.isResolved()
).on('error', (error) =>
  # It could already be resolved by an exception from bindWithFuture or bindWithOnException
  mainFuture.throw error unless mainFuture.isResolved()
)

mainFuture.wait()
wait futures