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.");
        });
}