>下载包含大量有关网页元数据的大型文件
>使用NPM事件流包从正则表达式拆分的文件元数据中创建流
>检查集合中的元数据是否匹配(我一直试图将每个网页的元数据流式传输到另一个函数来执行此操作)

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

作业集合包中的每个作业都已在异步函数中：

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.我想在Meteor.wrapAsync中包装最里面的searchWebPageMetaData()函数吗？ (当我输入时,想想我回答的是)
>流是否会缓慢以补偿数据库调用的缓慢？我的猜测不是,但我该怎么处理？

我花了很长时间了解Meteor的wrapAsync和bindEnvironment,但是无法将它们整合在一起并了解在哪里使用它们.

补充1

只是为了澄清,步骤是：

>下载文件;
>创建流;
>解压缩;
>将其拆分为单独的webPages – EventStream处理此问题
>将其发送给函数 – 不需要返回值;这可能是阻塞,只是一些搜索和数据库调用

我试图做这样的事情,除了我需要帮助的核心代码是在一个不同文件的函数中.以下代码中包含@ 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,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();
  }