CODEWITHSUNDEEP
node.jsmongodbasynchronousmongoose
Unmeow
Unmeow

Reputation: 37

Mongoose with async queue & waterfall

I aim to import large amount of data by Mongoose. As a newbie, I fail to setup the flow control properly with various mechanisms by async. Glad if someone could point to an appropriate solution. Thanks.

var async = require('async'),
    mongoose = require('mongoose');

mongoose.connect('mongodb://localhost/test');
var Cat = mongoose.model('Cat', { name: String });

// Imagine this is a huge array with a million items.
var content = ['aaa', 'bbb', 'ccc'];
var queries = [];
content.forEach(function(name) {
  queries.push(function(cb) {
    var obj = new Cat({ name: name });
    obj.save(function(err) {
      console.log("SAVED: " + name);
      console.log(err);
    });
    return true;
  });
});

// FAILED: async.parallel adds all content to db, 
// but it would exhaust the resource with too many parallel tasks.
async.parallel(queries, function(err, result) {
  if (err)
    return console.log(err);
  console.log(result);
});

// FAILED: save the first item but not the rest
async.waterfall(queries, function(err, result) {
  if (err)
    return console.log(err);
  console.log(result);
});

// FAILED: same as async.waterfall, async.queue saves the first item only
var q = async.queue(function(name, cb) {
  var obj = new Cat({ name: name });
  obj.save(function(err) {
    console.log("SAVED: " + name);
    console.log(err);
  });
})
q.push(content, function (err) {
  console.log('finished processing queue');
});

Upvotes: 2

Views: 4390

Answers (1)

robertklep
robertklep

Reputation: 203241

I think eachLimit or eachSeries fit your situation best:

var content = ['aaa', 'bbb', 'ccc'];
async.eachLimit(content, 10, function(name, done) {
  var obj = new Cat({ name : name });
  obj.save(done);
  // if you want to print some status info, use this instead:
  //
  // obj.save(function(err) {
  //   console.log("SAVED: " + name);
  //   console.log(err);
  //   done(err);
  // });
  //
}, function(err) {
  // handle any errors;
});

With eachLimit, you can run an X amount of queries 'in parallel' (10 in the example above) to speed things up without exhausting resources. eachSeries will wait for the previous save before it continues with the next, so effectively saving one object at a time.

Notice that with each*, you won't get a list with (saved) objects back (it's a bit of a fire-and-forget mechanism where you're not interested in the outcome, bar any errors). If you do want a list of saved objects in the end, you can use the equivalent map* functions: mapLimit and mapSeries.

Upvotes: 5

Related Questions