Node & MySQL: Can't end connection -> Async confusion

Question

I'm trying to write a Node program that populates my MySQL database with data from files I have on disk. I may or may not be going about this the right way, but it's working. What I'm having trouble with is understanding how I should be handling allowing asynchronous functions to finish before the connection to the DB is ended. Ultimately, I'll be reading lots of data files, and insert them into the database like I did below. I could just use readFileSync instead of the asynchronous version, but I need to get a better handle on asynchronous functions.

When I insert the wine categories below, it works fine since it's not using an asynchronous function. However, when I use readFile to get data from a file, I get an error that connection ended before any of the queries were executed:

connection.connect( function(err) {
    if(err) {
        console.log(err);
    }
});

// Take a table and the values, and insert a new row into a table
function insert_into( table, values ) {
    if( values instanceof Array ) {
        values = values.map( function( value ) {
            return '"' + value + '"';
        }).join(', ');
    } else {
        values = '"' + values + '"';
    }

    var statement = 'INSERT INTO ' + table + ' VALUES (NULL, ' + values + ')';
    connection.query( statement, function(err, rows, fields) {
      if (err) throw err;

      console.log( values + " successfully added.");
    });
};

// Populate the wine_categories table
var wine_categories = [
    'red', 'white', 'rose', 'sparkling', 'fortified'
];

// Works fine when used alone
wine_categories.forEach( function( element ) {
    insert_into( 'wine_categories', element );
});

// Populate the countries table
// connection.end() runs before this finishes its job
fs.readFile( countries, 'utf8', function (err, data) {
    if (err) {
        throw err;
    } else {
        var codes = Array.prototype.map.call( 
            data.split('\n'), function( country ) {
                return country.split('\t');
        });

        codes.forEach( function( country ) {
            if( country[1].length > 25 ) {
                country[1] = country[1].substring(0, 25);
            }
            insert_into( 'countries', country );
        });
    }
}); 

connection.end();

Obviously, connection.end() needs to happen after all of the inserts have completed, but I'm not sure how to handle that. I don't want it to be a callback for the readFile call because I'll ultimately have many of similar calls in this file.

How should I structure my code so that all of the queries execute and connection.end() runs when they're all finished? The answer is probably obvious to an asynchronous wiz...

Answer 1

Using promises it would be like this:

pool.getConnectionAsync().then(function(connection) {
    // Populate the wine_categories table
    var wine_categories = [
        'red', 'white', 'rose', 'sparkling', 'fortified'
    ];
    var wineQueries = wine_categories.map(function(wine){
        return insert_into(connection, "wine_categories", wine);
    });

    var countryQueries = fs.readFileAsync(countries, "utf-8").then(function(data) {
        return data.split("\n").map(function(country) {
            country = country.split("\t")[1];
            if (country.length > 25) {
                country = country.substring(0, 25);
            }
            return insert_into(connection, "countries", country);
        });
    });

    Promise.all(wineQueries.concat(countryQueries))
        .then(function() {
            console.log("all done");
        })
        .catch(function(e) {
            console.log("error", e);
        })
        .finally(function() {
            connection.release();
        })
});

Pre-requisite code for the above

var Promise = require("bluebird");
var fs = Promise.promisifyAll(require("fs"));
Promise.promisifyAll(require("mysql/lib/Connection").prototype);
var pool = Promise.promisifyAll(require("mysql").createPool({
    "user": "...",
    "password": "...",
    "database": "...",
    "host": "localhost",
    "port": 3306,
    "debug": false
}));

function insert_into(connection, table, values) {
    if( values instanceof Array ) {
        values = values.map(connection.escape, connection).join(', ');
    } else {
        values = connection.escape(values);
    }
    return connection
        .queryAsync('INSERT INTO ' + table + ' VALUES (NULL, ' + values + ')')
        .then(function() {
            console.log(values + " successfully added.");
        });
}

Answer 2

Assuming that insert_into is also asynchronous, you may want to use something like async.each to handle inserting your records. It has a convenient callback that will be called when all records are inserted, because only at that point do you want to close the connection:

async.each(codes, function(country, callback) {
  if ( country[1].length > 25 ) {
    country[1] = country[1].substring(0, 25);
  }
  insert_into( 'countries', country, callback ); // !! read below
}, function(err) {
  // TODO: handle any errors
  ...
  // Here, all countries are inserted.
  connection.end();
});

However, this means that insert_into should also be made to accept a callback (using the common Node convention function(err, result)) that will be called when the record has been inserted. In the code above, I'm using the callback provided by async directly, meaning that once your insert_into is done, it will call the async callback signaling that this iteration of each is done.

EDIT: you can rewrite insert_into so it looks like this:

function insert_into( table, values, callback ) {
  ...
  connection.query(..., function(err) {
    callback(err);
  });
}

Since you don't need the actual result from connection.query, you only have to pass err (instead of throwing it).

Tip: assuming that you're using node-mysql, you may want to take a look at the docs on how it can help you with escaping.