使用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

1 {
2     "name": "async-mysql",
3     "version": "1.0.0",
4     "main": "index.js",
5     "author": {
6         "name": "kazetotori/fxqn",
7         "email": "kakkouto98045@live.com"
8     },
9     "files": [
10         "lib",
11         "index.js"
12     ],
13     "dependencies": {
14         "mysql": "2.12.0",
15         "bluebird": "3.4.6"
16     }
17 }

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

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

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

1 const mysql = require('mysql');
2 const bluebird = require('bluebird');
3 const Promise = bluebird.Promise;
4 var $originConn = Symbol('originConn');
5 var $isPoolConn = Symbol('isPoolConn');
6 var $isAlive = Symbol('isAlive');
7
8 /**
9  * This function is the factory of the standard promise callback.
10  * @param {Function} resolve
11  * @param {Function} reject
12  * @return {Function} The standard promise callback.
13  */
14 function promiseFn(resolve, reject) {
15     return (err, rst) => {
16         if (err) reject(err);
17         else resolve(rst);
18     }
19 }
20
21
22
23 /**
24  * Connection is the class that contains functions that each returns promise.
25  * These functions are just converted from a Mysql.Connection Object.
26  */
27 class Connection {
28
29     /**
30      * Constructor, initialize the connection object.
31      * @param {Object} config The configuration of the connection.
32      * @param {Boolean} isPoolConn Orders the connection is in a pool or not.
33      */
34     constructor(config, isPoolConn = false) {
35         if (config.query)
36             this[$originConn] = config;
37         else
38             this[$originConn] = mysql.createConnection(config);
39         this[$isPoolConn] = isPoolConn;
40         this[$isAlive] = true;
41     }
42
43     /**
44      * Connection config
45      */
46     get config() { return this[$originConn].config; }
47
48     /**
49      * Orders the connection is in a pool or not.
50      */
51     get isPoolConnection() { return this[$isPoolConn]; }
52
53     /**
54      * Orders the connection is destroyed or not.
55      */
56     get isAlive() { return this[$isAlive]; }
57
58     /**
59      * Orders the threadId of the connection.
60      */
61     get threadId() { return this[$originConn].threadId; }
62
63     /**
64      * Add listener of this connection.
65      */
66     get on() { return this[$originConn].on; };
67
68     /**
69      * Ternimate the connection immediately whether there's any query in quene or not.
70      */
71     destroy() {
72         return new Promise((resolve, reject) => {
73             this[$originConn].destroy();
74             this[$isAlive] = false;
75             resolve();
76         });
77     }
78
79     /**
80      * Ternimate the connection. This function will ternimate the connection after any query being complete.
81      */
82     end() {
83         return new Promise((resolve, reject) => {
84             this[$originConn].end(promiseFn(resolve, reject))
85         })
86             .then(() => {
87                 this[$isAlive] = false;
88             })
89     }
90
91     /**
92      * Execute sql command with parameters.
93      * @param {String} cmd The sql command would be executed.
94      * @param {Array} params Parameters.
95      * @return {Promise<any>} The sql result.
96      */
97     query(cmd, params) {
98         return new Promise((resolve, reject) => {
99             let conn = this[$originConn];
100             let args = [cmd];
101             let callback = promiseFn(resolve, reject);
102             if (params)
103                 args.push(params);
104             args.push(callback);
105             conn.query(...args);
106         });
107     }
108
109     /**
110      * Begin transaction of the connection. Following queries would not be useful until the function commit or rollback called.
111      * @return {Promise<undefined>}
112      */
113     beginTran() {
114         return new Promise((resolve, reject) => {
115             let conn = this[$originConn];
116             conn.beginTransaction(promiseFn(resolve, reject));
117         });
118     }
119
120     /**
121      * Commit a transaction.
122      * @return {Promise<undefined>}
123      */
124     commit() {
125         return new Promise((resolve, reject) => {
126             let conn = this[$originConn];
127             conn.commit((err) => {
128                 if (err) this.rollback().then(() => reject(err));
129                 else resolve();
130             })
131         });
132     }
133
134     /**
135      * Rollback a transaction
136      * @return {Promise<undefined>}
137      */
138     rollback() {
139         return new Promise((resolve, reject) => {
140             let conn = this[$originConn];
141             conn.rollback(() => resolve());
142         });
143     }
144 }
145
146
147 /**
148  * PoolConnection is the class extending from Connection.
149  * Any object of this class is the connection in a connection pool.
150  */
151 class PoolConnection extends Connection {
152     constructor(originConn) {
153         super(originConn, true);
154     }
155
156     /**
157      * Release the connection and put it back to the pool.
158      * @return {Promise<undefined>}
159      */
160     release() {
161         return new Promise((resolve, reject) => {
162             this[$originConn].release();
163             resolve();
164         });
165     }
166 }
167
168
169
170 module.exports.Connection = Connection;
171 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();