Adding JS functions to a parallel queue

Question

I have this code, I want to run all the database inserts and then disconnect from the database at the end of the script. The problem is that I need to execute the disconnect callback after the loop ends, but it's non-deterministic. So I would need to create a function for each of the loops, and then only after all those functions have completed, call the function to disconnect from mongoose/mongoDB.

Does anyone see the problem that I have with this code?

//this code is close, but not quite, since it will disconnect from mongo before doing very many inserts/saves

var mongoose = require('mongoose')
    , Admin = mongoose.mongo.Admin;

var UserModel = require('../models/UserModel');
UserModel.registerSchema(mongoose);


var fs = require('fs');
var parsedJSON = JSON.parse(fs.readFileSync('../dummy_data/dummy_user_data', 'utf8'));


var system_db = mongoose.connect('mongodb://localhost:27017/local_dev_db');


function insertUsers(callback){


    parsedJSON.forEach(function (item, index) {

        var User = UserModel.getNewUser(system_db);

        var user = new User({
            username: item.username,
            password: item.password,
            address: item.address,
            phone: item.phone,
            email: item.email,
            gender: item.gender,
            about: item.about,
            latitude: item.latitude,
            longitude: item.longitude
        });

        user.save(function (err, result) {
            if (err) {
                console.log("error in player save method:", err);
            }
            console.log(index);
            if (result) {
                //console.log('Added!', result);
            }
        });

    });
    callback();
}


function disconnect(){
    mongoose.disconnect();
    mongoose.connection
        .close(function () {

            console
                .log('Mongoose connection disconnected');

            process.exit(0);

        });
}

insertUsers(disconnect);

So the solution would be to use the async.parallel library, and only after all the functions are done, call a final callback function. However, how can I do that programmatically instead of putting a known group of functions into the async.parallel codeblock, I need to put an unknown number of functions into the aysnc.parallel codeblock. Ya seen?

Answer 1

Definitely can be done. Just dynamically add your functions to an array that will be passed to async.parallel:

var fs = require('fs');
var async = require('async');

var mongoose = require('mongoose'),
    Admin = mongoose.mongo.Admin;

var UserModel = require('../models/UserModel');
UserModel.registerSchema(mongoose);

var parsedJSON = JSON.parse(fs.readFileSync('../dummy_data/dummy_user_data', 'utf8'));

var system_db = mongoose.connect('mongodb://localhost:27017/local_dev_db');

var insertFunctions = [];

parsedJSON.forEach(function (item, index) {

    insertFunctions.push(function(callback) {

        var User = UserModel.getNewUser(system_db);

        var user = new User({
            username: item.username,
            password: item.password,
            address: item.address,
            phone: item.phone,
            email: item.email,
            gender: item.gender,
            about: item.about,
            latitude: item.latitude,
            longitude: item.longitude
        });

        user.save(function (err, result) {

            if (err) {
                console.log("error in player save method:", err);
                callback(err);
                return;
            }

             callback(null, result);

        });
    });
});

function disconnect() {

    mongoose.disconnect();
    mongoose.connection
        .close(function () {

            console
                .log('Mongoose connection disconnected');

            process.exit(0);
        });
}

//First parameter is the array of functions to run in parallel,
// second parameter is the callback function
async.parallel(insertFunctions, disconnect);

Answer 2

In this particular case, async.each() actually fits better than async.parallel(). async.each() is parallel while async.eachSeries() is not. Here is the code:

function insertUsers(callback){
  var User = UserModel.getNewUser(system_db);

  async.each(parsedJSON, function (item, eachCb) {
    var user = new User({
        username: item.username,
        password: item.password,
        address: item.address,
        phone: item.phone,
        email: item.email,
        gender: item.gender,
        about: item.about,
        latitude: item.latitude,
        longitude: item.longitude
    });

    user.save(function (err, result) {
        if (err) {
            console.log("error in player save method:", err);
        }
        console.log(index);
        if (result) {
            //console.log('Added!', result);
        }
        eachCb();  // if calling with eachCb(err), async.each()
                   // will not continue the rest of the items
                   // in case of error occurs
    });

  }, function(err) {
    callback();    // done with all user.save() calls
  });
}