使用es6特性封装async-mysql

node.js的mysql模块本身没有提供返回promise的函数，即是说都是用的回调函数，那么对于我们使用async函数是很不方便的一件事。node.js有一个mysql封装库叫mysql-promise，这个库提供使用函数拼凑sql语句，但我比较想用原生的sql语句，原因在于mysql对于query函数封装得比较完美，能对应select,delete,update,insert返回不同的结果，比如update和delete返回affectRows，select返回查询结果这样，再加上使用参数化的sql语句能防止sql注入，所以封装了一下mysql的npm包。

首先是package.json，对于Promise，建议使用bluebird.Promise，然后自然是mysql

{
    "name": "async-mysql",
    "version": "1.0.0",
    "main": "index.js",
    "author": {
        "name": "kazetotori/fxqn",
        "email": "kakkouto98045@live.com"
    },
    "files": [
        "lib",
        "index.js"
    ],
    "dependencies": {
        "mysql": "2.12.0",
        "bluebird": "3.4.6"
    }
}

这个库入口文件为index.js，这里仅仅作为一个导出文件使用，没有任何代码

module.exports.Connection = require('./lib/Connection').Connection;
module.exports.Pool = require('./lib/Pool').Pool;

首先来实现一下单个连接的各个函数，使用es6的class关键字和Symbol封装，比传统的构造函数更加直观

const mysql = require('mysql');
const bluebird = require('bluebird');
const Promise = bluebird.Promise;
var $originConn = Symbol('originConn');
var $isPoolConn = Symbol('isPoolConn');
var $isAlive = Symbol('isAlive');

/**
 * This function is the factory of the standard promise callback.
 * @param {Function} resolve
 * @param {Function} reject
 * @return {Function} The standard promise callback.
 */
function promiseFn(resolve, reject) {
    return (err, rst) => {
        if (err) reject(err);
        else resolve(rst);
    }
}



/**
 * Connection is the class that contains functions that each returns promise.
 * These functions are just converted from a Mysql.Connection Object.
 */
class Connection {

    /**
     * Constructor, initialize the connection object.
     * @param {Object} config The configuration of the connection.
     * @param {Boolean} isPoolConn Orders the connection is in a pool or not.
     */
    constructor(config, isPoolConn = false) {
        if (config.query)
            this[$originConn] = config;
        else
            this[$originConn] = mysql.createConnection(config);
        this[$isPoolConn] = isPoolConn;
        this[$isAlive] = true;
    }

    /**
     * Connection config
     */
    get config() { return this[$originConn].config; }

    /**
     * Orders the connection is in a pool or not.
     */
    get isPoolConnection() { return this[$isPoolConn]; }

    /**
     * Orders the connection is destroyed or not.
     */
    get isAlive() { return this[$isAlive]; }

    /**
     * Orders the threadId of the connection.
     */
    get threadId() { return this[$originConn].threadId; }

    /**
     * Add listener of this connection.
     */
    get on() { return this[$originConn].on; };

    /**
     * Ternimate the connection immediately whether there's any query in quene or not.
     */
    destroy() {
        return new Promise((resolve, reject) => {
            this[$originConn].destroy();
            this[$isAlive] = false;
            resolve();
        });
    }

    /**
     * Ternimate the connection. This function will ternimate the connection after any query being complete.
     */
    end() {
        return new Promise((resolve, reject) => {
            this[$originConn].end(promiseFn(resolve, reject))
        })
            .then(() => {
                this[$isAlive] = false;
            })
    }

    /**
     * Execute sql command with parameters.
     * @param {String} cmd The sql command would be executed.
     * @param {Array} params Parameters.
     * @return {Promise<any>} The sql result.
     */
    query(cmd, params) {
        return new Promise((resolve, reject) => {
            let conn = this[$originConn];
            let args = [cmd];
            let callback = promiseFn(resolve, reject);
            if (params)
                args.push(params);
            args.push(callback);
            conn.query(...args);
        });
    }

    /**
     * Begin transaction of the connection. Following queries would not be useful until the function commit or rollback called.
     * @return {Promise<undefined>}
     */
    beginTran() {
        return new Promise((resolve, reject) => {
            let conn = this[$originConn];
            conn.beginTransaction(promiseFn(resolve, reject));
        });
    }

    /**
     * Commit a transaction.
     * @return {Promise<undefined>}
     */
    commit() {
        return new Promise((resolve, reject) => {
            let conn = this[$originConn];
            conn.commit((err) => {
                if (err) this.rollback().then(() => reject(err));
                else resolve();
            })
        });
    }

    /**
     * Rollback a transaction
     * @return {Promise<undefined>}
     */
    rollback() {
        return new Promise((resolve, reject) => {
            let conn = this[$originConn];
            conn.rollback(() => resolve());
        });
    }
}


/**
 * PoolConnection is the class extending from Connection.
 * Any object of this class is the connection in a connection pool.
 */
class PoolConnection extends Connection {
    constructor(originConn) {
        super(originConn, true);
    }

    /**
     * Release the connection and put it back to the pool.
     * @return {Promise<undefined>}
     */
    release() {
        return new Promise((resolve, reject) => {
            this[$originConn].release();
            resolve();
        });
    }
}



module.exports.Connection = Connection;
module.exports.PoolConnection = PoolConnection;

然后是连接池的部分

const Promise = require('bluebird').Promise;
const mysql = require('mysql');
const PoolConnection = require('./Connection').PoolConnection;
var $originPool = Symbol('originPool');
var $isAlive = Symbol('isAlive');


/**
 * Pool is the class that contains functions each returns promise.
 * These functions are just converted from the Mysql.Pool object.
 */
class Pool {

    /**
     * Constructor, initialize the pool.
     * @param {Object} config The pool config.
     */
    constructor(config) {
        this[$originPool] = mysql.createPool(config);
        this[$isAlive] = true;
    }

    /**
     * Orders the pool config.
     */
    get config() { return this[$originPool].config; }

    /**
     * Orders the pool is destroyed or not.
     */
    get isAlive() { return this[$isAlive]; }

    /**
     * Add listener to the pool.
     */
    get on() { return this[$originPool].on; }

    /**
     * Get a connection object from the pool.
     * @return {Promise<PoolConnection>}
     */
    getConn() {
        return new Promise((resolve, reject) => {
            this[$originPool].getConnection((err, originConn) => {
                if (err)
                    return reject(err);
                let conn = new PoolConnection(originConn);
                resolve(conn);
            });
        });
    }

    /**
     * Ternimate the pool. This function would ternimate the pool after any query being complete.
     */
    end() {
        return new Promise((resolve, reject) => {
            this[$originPool].end((err) => {
                if (err)
                    return reject(err);
                this[$isAlive] = false;
                resolve();
            })
        });
    }

    /**
     * Use a connection to query a sql command with parameters.
     * @param {String} cmd The sql command would be executed.
     * @param {Array} params Parameters.
     * @return {Promise<any>} 
     */
    query(cmd, params) {
        return new Promise((resolve, reject) => {
            let args = [cmd];
            let callback = (err, rst) => {
                if (err) reject(err);
                else resolve(rst);
            }
            if (params)
                args.push(params);
            args.push(callback);
            this[$originPool].query(...args);
        });
    }
}



module.exports.Pool = Pool;

最后加一个config，便于智能提示

var $host = Symbol('host');
var $port = Symbol('port');
var $localAddr = Symbol('localAddr');
var $socketPath = Symbol('socketPath');
var $user = Symbol('user');
var $pwd = Symbol('pwd');
var $db = Symbol('db');
var $charset = Symbol('charset');
var $timezone = Symbol('timezone');
var $connTimeout = Symbol('connTimeout');
var $stringifyObjs = Symbol('stringifyObjs');
var $typeCast = Symbol('typeCast');
var $queryFormat = Symbol('queryFormat');
var $supportBigNumbers = Symbol('supportBigNumbers');
var $bigNumberStrings = Symbol('bigNumberStrings');
var $dateStrings = Symbol('dateStrings');
var $debug = Symbol('debug');
var $trace = Symbol('trace');
var $multiStmts = Symbol('multipleStatements');
var $flags = Symbol('flags');
var $ssl = Symbol('ssl');

class MysqlConfig {
    constructor(config) {
        for (let k in config)
            this[k] = config[k];
    }

    get host() { return this[$host] }
    set host(val) { this[$host] = val }

    get port() { return this[$port] }
    set port(val) { this[$port] = val }

    get localAddress() { return this[$localAddr] }
    set localAddress(val) { this[$localAddr] = val }

    get socketPath() { return this[$socketPath] }
    set socketPath(val) { this[$socketPath] = val }

    get user() { return this[$user] }
    set user(val) { this[$user] = val }

    get password() { return this[$pwd] }
    set password(val) { this[$pwd] = val }

    get database() { return this[$db] }
    set database(val) { this[$db] = val }

    get charset() { return this[$charset] }
    set charset(val) { this[$charset] = val }

    get timezone() { return this[$timezone] }
    set timezone(val) { this[$timezone] = val }

    get connectTimeout() { return this[$connTimeout] }
    set connectTimeout(val) { this[$connTimeout] = val }

    get stringifyObjects() { return this[$stringifyObjs] }
    set stringifyObjects(val) { this[$stringifyObjs] = val }

    get typeCast() { return this[$typeCast] }
    set typeCast() { this[$typeCast] = val }

    get queryFormat() { return this[$queryFormat] }
    set queryFormat(val) { this[$queryFormat] = val }

    get supportBigNumbers() { return this[$supportBigNumbers] }
    set supportBigNumbers(val) { this[$supportBigNumbers] = val }

    get bigNumberStrings() { return this[$bigNumberStrings] }
    set bigNumberStrings(val) { this[$bigNumberStrings] = val }

    get dateStrings() { return this[$dateStrings] }
    set dateStrings(val) { this[$dateStrings] = val }

    get debug() { return this[$debug] }
    set debug(val) { this[$debug] = val }

    get trace() { return this[$trace] }
    set trace(val) { this[$trace] = val }

    get multipleStatements() { return this[$multiStmts] }
    set multipleStatements(val) { this[$multiStmts] = val }

    get flags() { return this[$flags] }
    set flags(val) { this[$flags] = val }

    get ssl() { return this[$ssl] }
    set ssl(val) { this[$ssl] = val }
}

module.exports.MysqlConfig = MysqlConfig;

测试代码

//Use this test.js need node version is higher than 7.0.0 .
//And need the node arg "--harmony".

const config = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "database": "testDB",
    "charset": "UTF8_GENERAL_CI",
    "timezone": "local",
    "connectTimeout": 10000,
    "connectionLimit": 10
};
const Pool = require('./lib/Pool').Pool;
const Connection = require('./lib/Connection').Connection;
var pool = new Pool(config);
var conn = new Connection(config);

async function poolTest() {

    //pool.query()
    let result = await pool.query('SELECT * FROM tbltest WHERE name=?', ['wr']);
    console.log(result);

    //pool.getConn();
    let poolConn = await pool.getConn();
    console.log(poolConn.isPoolConnection);
    result = await poolConn.query('SELECT * FROM tbltest WHERE name=?', ['zs']);
    console.log(result);

    await pool.end();
    console.log(pool.isAlive);
}

async function connTest() {
    let rst = await conn.query('SELECT * FROM tbltest WHERE name=?', ['ls']);
    console.log(rst);
    await conn.beginTran();
    let count = (await conn.query('SELECT COUNT(*) FROM tbltest WHERE name=?', ['??']))[0]['COUNT(*)'];
    console.log(count);
    await conn.query('INSERT INTO tbltest(name) VALUES(?)', ['zhangsan']);
    if (count > 0) {
        await conn.commit();
        console.log('commit');
    }
    else {
        await conn.rollback();
        console.log('rollback');
    }

    rst = await conn.query('SELECT * FROM tbltest');
    console.log(rst);
}

poolTest();
connTest();