<p>This is quite tricky if you want to handle all errors correctly. So one should ask themself, what to do if: you code throws an exception, or <code>error</code> event handler is called. You want that errors propagate correctly, that is, are thrown as an exception in the fiber calling streaming code. I implemented something like this for one of our <a href="https://github.com/peerlibrary/peerlibrary/blob/development/server/jobs/arxiv.coffee#L222" rel="nofollow">job-collecton</a> tasks, for extracting <a href="https://github.com/peerlibrary/meteor-tar" rel="nofollow">tar files</a>.</p>

<p>First you need some helper functions:</p>

<pre><code>bindWithFuture = (futures, mainFuture, fun, self) -&gt;
  wrapped = (args...) -&gt;
    future = new Future()

    if mainFuture
      future.resolve (error, value) -&gt;
        # 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) -&gt;
  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) -&gt;
      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
</code></pre>

<p>And then you can run something like:</p>

<pre><code>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) =&gt;
  Meteor.bindEnvironment f, (error) =&gt;
    mainFuture.throw error unless mainFuture.isResolved()

onWebpageMetaData = (metaData, callback) =&gt;
  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) -&gt;
    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', =&gt;
  # It could already be resolved by an exception from bindWithFuture or bindWithOnException
  mainFuture.return() unless mainFuture.isResolved()
).on('error', (error) =&gt;
  # It could already be resolved by an exception from bindWithFuture or bindWithOnException
  mainFuture.throw error unless mainFuture.isResolved()
)

mainFuture.wait()
wait futures
</code></pre>
Meteor: How do I stream and parse a large file to an async Node function?
<p>I'm using the <a href="https://github.com/vsivsi/meteor-job-collection" rel="nofollow">job-collection</a> package to do the following:</p>

<ol>
<li>Download a large file with a bunch of metadata about webpages</li>
<li>Create a stream from the file metadata that is split by a regex using the NPM <code>event-stream</code> package</li>
<li>Check if there is a match of the metadata in a collection (I've been attempting to stream each webpage's metadata to another function to do this) </li>
</ol>

<p>The file is too large to buffer, so streaming is required. <a href="https://s3.amazonaws.com/ja-common-crawl/exampleWatFile.wat.gz" rel="nofollow">Here is a small file with a few examples of the metadata </a> if you wish to try this.</p>

<p>Each job from the <code>job-collection</code> package is already inside an async function: </p>

<pre><code>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
}
</code></pre>

<ul>
<li>Are there better ways to structure this?  Am I on the right track?</li>
<li>Where to put <code>Meteor.bindEnvironment</code>? - do I I bind the environment for each time I pass to <code>searchWebPageMetaData()</code>?  Do I need to expressly use fibers here?</li>
<li>The stream stops when running this if I run it to <code>process.stdout</code>. Am I supposed to put the stream into one of Meteor's wrap</li>
<li>I'm aware of <code>Meteor.wrapAsync</code>.  Do I want to wrap the innermost <code>searchWebPageMetaData()</code> function in <code>Meteor.wrapAsync</code>? (think I'm answering this yes as I type)</li>
<li>Will the stream slow to compensate for the slowness of the DB calls?   My guess is no but how do I deal with that?</li>
</ul>

<p>I've spent quite a while learning about Meteor's <code>wrapAsync</code>, and <code>bindEnvironment</code> but having trouble bringing it all together and understanding where to use them.</p>

<p><strong>SUPPLEMENT 1</strong></p>

<p>Just to clarify, the steps are:</p>

<ol>
<li>Download file;</li>
<li>Create stream;</li>
<li>unzip it;</li>
<li>split it into individual webPages - EventStream handles this</li>
<li>send it to a function - don't need return values; this could be blocking, it's just some searching and database call</li>
</ol>

<p>I was trying to do something like this, except the core code I need help with was in a function in a different file.  The following code has most of @electric-jesus' answer in there.</p>

<pre><code>   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();
  }
</code></pre>