Node.js Streams on(end) completing before asynchronous on(readable) completed

Question

I'm using the Node.js request library along with node-feedparser to capture a feed and it's posts into a MongoDB database using Mongoose.

I'm storing the post information into a posts collection and the feed information into a feeds collection, but I need to store the post._id within an array in the feeds collection called feeds._post.

The problem I'm having is using the stream interface, the feedparser.on('end') is called before all of the feedparser.on('readable')'s async calls to the DB have completed, therefore I end up with say 15 posts in the Post collection and only 11 post._id's in the Feed._post array.

I understand that if this was just plain JavaScript, I could use async to ensure that the .on('end') waits for all the .on('readable')'s to complete, but how do I go about this with streams?

Thanks in advance.

db.Feed.findById(feedid, function(error, feed) {

// request.on('response') -> this.pipe(feedparser)

  feedparser.on('readable', function() {
    var post;
    while (null !== (post = this.read())) {
      db.Post.create({ /* post details */ }, function(err, post) {
        feed._post.push(post);
      });
    }
  });

  feedparser.on('end', function() {
    feed.save();
  });

});

Answer 1

You'll need to keep track of a counter and a boolean. Increment the counter when the "readable" event first fires and decrement the counter when you're finished saving it to the database. The boolean starts out in one state and is switched when the "end" event fires. For example:

 var processing = 0, done = false;

 var finished = function(){
     if(processing === 0 && done){
         feed.save();
         // ... other stuff
     }
 };

 feedparser.on("readable", function(){
     processing++;
     doStuff(something, function(){
         // something asynchronous         
         processing--;
         finished();
     });
 });

 feedparser.on("end", function(){
     done = true;
     finished();
 });