function Connection(options) {
Events.EventEmitter.call(this);
this.config = options.config;
this._socket = options.socket;
this._protocol = new Protocol({config: this.config, connection: this});
this._connectCalled = false;
this.state = 'disconnected';
this.threadId = null;
}n/a
function Pool(options) {
EventEmitter.call(this);
this.config = options.config;
this.config.connectionConfig.pool = this;
this._acquiringConnections = [];
this._allConnections = [];
this._freeConnections = [];
this._connectionQueue = [];
this._closed = false;
}n/a
function PoolCluster(config) {
EventEmitter.call(this);
config = config || {};
this._canRetry = typeof config.canRetry === 'undefined' ? true : config.canRetry;
this._defaultSelector = config.defaultSelector || 'RR';
this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
this._restoreNodeTimeout = config.restoreNodeTimeout || 0;
this._closed = false;
this._findCaches = Object.create(null);
this._lastId = 0;
this._namespaces = Object.create(null);
this._nodes = Object.create(null);
}n/a
function Query(options,
callback) {
Sequence.call(this, options, callback);
this.sql = options.sql;
this.values = options.values;
this.typeCast = (options.typeCast === undefined)
? true
: options.typeCast;
this.nestTables = options.nestTables || false;
this._resultSet = null;
this._results = [];
this._fields = [];
this._index = 0;
this._loadError = null;
}n/a
function createConnection(config) {
var Connection = loadClass('Connection');
var ConnectionConfig = loadClass('ConnectionConfig');
return new Connection({config: new ConnectionConfig(config)});
}...
This is a node.js driver for mysql. It is written in JavaScript, does not
require compiling, and is 100% MIT licensed.
Here is an example on how to use it:
```js
var mysql = require('mysql');
var connection = mysql.createConnection({
host : 'localhost',
user : 'me',
password : 'secret',
database : 'my_db'
});
connection.connect();
...function createPool(config) {
var Pool = loadClass('Pool');
var PoolConfig = loadClass('PoolConfig');
return new Pool({config: new PoolConfig(config)});
}...
```
Unlike `end()` the `destroy()` method does not take a callback argument.
## Pooling connections
Rather than creating and managing connections one-by-one, this module also
provides built-in connection pooling using `mysql.createPool(config)`.
[Read more about connection pooling](https://en.wikipedia.org/wiki/Connection_pool).
Use pool directly.
```js
var mysql = require('mysql');
var pool = mysql.createPool({
connectionLimit : 10,
...function createPoolCluster(config) {
var PoolCluster = loadClass('PoolCluster');
return new PoolCluster(config);
}...
## PoolCluster
PoolCluster provides multiple hosts connection. (group & retry & selector)
```js
// create
var poolCluster = mysql.createPoolCluster();
// add configurations (the config is a pool config object)
poolCluster.add(config); // add configuration with automatic name
poolCluster.add('MASTER', masterConfig); // add a named configuration
poolCluster.add('SLAVE1', slave1Config);
poolCluster.add('SLAVE2', slave2Config);
...function createQuery(sql,
values,
callback) {
var Connection = loadClass('Connection');
return Connection.createQuery(sql, values, callback);
}...
(function () {
'use strict';
var local;
local = require('mysql');
local.Connection = local.createConnection({}).constructor;
local.Pool = local.createPool({}).constructor;
local.PoolCluster = local.createPoolCluster({}).constructor;
local.Query = local.createQuery({}).constructor;
module.exports = local;
}());
...function escape(value,
stringifyObjects,
timeZone) {
var SqlString = loadClass('SqlString');
return SqlString.escape(value, stringifyObjects, timeZone);
}...
);
```
## Escaping query values
In order to avoid SQL Injection attacks, you should always escape any user
provided data before using it inside a SQL query. You can do so using the
`mysql.escape()`, `connection.escape()` or `pool.escape()&#x\
60; methods:
```js
var userId = 'some user provided value';
var sql = 'SELECT * FROM users WHERE id = ' + connection.escape(userId);
connection.query(sql, function(err, results) {
// ...
});
...function escapeId(value,
forbidQualified) {
var SqlString = loadClass('SqlString');
return SqlString.escapeId(value, forbidQualified);
}...
console.log(query); // SELECT * FROM posts WHERE title='Hello MySQL'
```
## Escaping query identifiers
If you can't trust an SQL identifier (database / table / column name) because it is
provided by a user, you should escape it with `mysql.escapeId(identifier)`,
`connection.escapeId(identifier)` or `pool.escapeId(identifier)` like this:
```js
var sorter = 'date';
var sql = 'SELECT * FROM posts ORDER BY ' + connection.escapeId(sorter);
connection.query(sql, function(err, results) {
// ...
...function format(sql,
values,
stringifyObjects,
timeZone) {
var SqlString = loadClass('SqlString');
return SqlString.format(sql, values, stringifyObjects, timeZone);
}...
### Preparing Queries
You can use mysql.format to prepare a query with multiple insertion points, utilizing the proper escaping for ids and values. A \
simple example of this follows:
```js
var sql = "SELECT * FROM ?? WHERE ?? = ?";
var inserts = ['users', 'id', userId];
sql = mysql.format(sql, inserts);
```
Following this you then have a valid, escaped query that you can then send to the database safely. This is useful if you are loo\
king to prepare the query before actually sending it to the database. As mysql.format is exposed from SqlString.format you also \
have the option (but are not required) to pass in stringifyObject and timezone, allowing you provide a custom means of turning o\
bjects into strings, as well as a location-specific/timezone-aware Date.
### Custom format
If you prefer to have another type of query escape format, there's a connection configuration option you can use to define \
a custom format function. You can access the connection object if you want to use the built-in `.escape()` or any othe\
r connection function.
...function Connection(options) {
Events.EventEmitter.call(this);
this.config = options.config;
this._socket = options.socket;
this._protocol = new Protocol({config: this.config, connection: this});
this._connectCalled = false;
this.state = 'disconnected';
this.threadId = null;
}n/a
function createQuery(sql,
values,
callback) {
if (sql instanceof Query) {
return sql;
}
var cb = bindToCurrentDomain(callback);
var options = {};
if (typeof sql === 'function') {
cb = bindToCurrentDomain(sql);
return new Query(options, cb);
}
if (typeof sql === 'object') {
for (var prop in sql) {
options[prop] = sql[prop];
}
if (typeof values === 'function') {
cb = bindToCurrentDomain(values);
} else if (values !== undefined) {
options.values = values;
}
return new Query(options, cb);
}
options.sql = sql;
options.values = values;
if (typeof values === 'function') {
cb = bindToCurrentDomain(values);
options.values = undefined;
}
if (cb === undefined && callback !== undefined) {
throw new TypeError('argument callback must be a function when provided');
}
return new Query(options, cb);
}...
(function () {
'use strict';
var local;
local = require('mysql');
local.Connection = local.createConnection({}).constructor;
local.Pool = local.createPool({}).constructor;
local.PoolCluster = local.createPoolCluster({}).constructor;
local.Query = local.createQuery({}).constructor;
module.exports = local;
}());
...function EventEmitter() {
EventEmitter.init.call(this);
}n/a
function beginTransaction(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
options = options || {};
options.sql = 'START TRANSACTION';
options.values = null;
return this.query(options, callback);
}...
```
## Transactions
Simple transaction support is available at the connection level:
```js
connection.beginTransaction(function(err) {
if (err) { throw err; }
connection.query('INSERT INTO posts SET title=?', title, function(err, result) {
if (err) {
return connection.rollback(function() {
throw err;
});
}
...function changeUser(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
this._implyConnect();
var charsetNumber = (options.charset)
? ConnectionConfig.getCharsetNumber(options.charset)
: this.config.charsetNumber;
return this._protocol.changeUser({
user : options.user || this.config.user,
password : options.password || this.config.password,
database : options.database || this.config.database,
timeout : options.timeout,
charsetNumber : charsetNumber,
currentConfig : this.config
}, bindToCurrentDomain(callback));
}...
## Switching users and altering connection state
MySQL offers a changeUser command that allows you to alter the current user and
other aspects of the connection without shutting down the underlying socket:
```js
connection.changeUser({user : 'john'}, function(err) {
if (err) throw err;
});
```
The available options for this feature are:
* `user`: The name of the new user (defaults to the previous one).
...function commit(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
options = options || {};
options.sql = 'COMMIT';
options.values = null;
return this.query(options, callback);
}...
connection.query('INSERT INTO log SET data=?', log, function(err, result) {
if (err) {
return connection.rollback(function() {
throw err;
});
}
connection.commit(function(err) {
if (err) {
return connection.rollback(function() {
throw err;
});
}
console.log('success!');
});
...function connect(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
if (!this._connectCalled) {
this._connectCalled = true;
// Connect either via a UNIX domain socket or a TCP socket.
this._socket = (this.config.socketPath)
? Net.createConnection(this.config.socketPath)
: Net.createConnection(this.config.port, this.config.host);
// Connect socket to connection domain
if (Events.usingDomains) {
if (this._socket.domain) {
this._socket.domain.remove(this._socket);
}
if (this.domain) {
this.domain.add(this._socket);
}
}
var connection = this;
this._protocol.on('data', function(data) {
connection._socket.write(data);
});
this._socket.on('data', function(data) {
connection._protocol.write(data);
});
this._protocol.on('end', function() {
connection._socket.end();
});
this._socket.on('end', function(err) {
connection._protocol.end();
});
this._socket.on('error', this._handleNetworkError.bind(this));
this._socket.on('connect', this._handleProtocolConnect.bind(this));
this._protocol.on('handshake', this._handleProtocolHandshake.bind(this));
this._protocol.on('unhandledError', this._handleProtocolError.bind(this));
this._protocol.on('drain', this._handleProtocolDrain.bind(this));
this._protocol.on('end', this._handleProtocolEnd.bind(this));
this._protocol.on('enqueue', this._handleProtocolEnqueue.bind(this));
if (this.config.connectTimeout) {
var handleConnectTimeout = this._handleConnectTimeout.bind(this);
this._socket.setTimeout(this.config.connectTimeout, handleConnectTimeout);
this._socket.once('connect', function() {
this.setTimeout(0, handleConnectTimeout);
});
}
}
this._protocol.handshake(options, bindToCurrentDomain(callback));
}...
var connection = mysql.createConnection({
host : 'localhost',
user : 'me',
password : 'secret',
database : 'my_db'
});
connection.connect();
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
...destroy = function () {
this.state = 'disconnected';
this._implyConnect();
this._socket.destroy();
this._protocol.destroy();
}...
An alternative way to end the connection is to call the `destroy()` method.
This will cause an immediate termination of the underlying socket.
Additionally `destroy()` guarantees that no more events or callbacks will be
triggered for the connection.
```js
connection.destroy();
```
Unlike `end()` the `destroy()` method does not take a callback argument.
## Pooling connections
Rather than creating and managing connections one-by-one, this module also
...function end(options,
callback) {
var cb = callback;
var opts = options;
if (!callback && typeof options === 'function') {
cb = options;
opts = null;
}
// create custom options reference
opts = Object.create(opts || null);
if (opts.timeout === undefined) {
// default timeout of 30 seconds
opts.timeout = 30000;
}
this._implyConnect();
this._protocol.quit(opts, bindToCurrentDomain(cb));
}...
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
connection.end();
```
From this example, you can learn the following:
* Every method you invoke on a connection is queued and executed in sequence.
* Closing the connection is done using `end()` which makes sure all remaining
queries are executed before sending a quit packet to the mysql server.
...escape = function (value) {
return SqlString.escape(value, false, this.config.timezone);
}...
);
```
## Escaping query values
In order to avoid SQL Injection attacks, you should always escape any user
provided data before using it inside a SQL query. You can do so using the
`mysql.escape()`, `connection.escape()` or `pool.escape()&#x\
60; methods:
```js
var userId = 'some user provided value';
var sql = 'SELECT * FROM users WHERE id = ' + connection.escape(userId);
connection.query(sql, function(err, results) {
// ...
});
...function escapeId(value) {
return SqlString.escapeId(value, false);
}...
console.log(query); // SELECT * FROM posts WHERE title='Hello MySQL'
```
## Escaping query identifiers
If you can't trust an SQL identifier (database / table / column name) because it is
provided by a user, you should escape it with `mysql.escapeId(identifier)`,
`connection.escapeId(identifier)` or `pool.escapeId(identifier)` like this:
```js
var sorter = 'date';
var sql = 'SELECT * FROM posts ORDER BY ' + connection.escapeId(sorter);
connection.query(sql, function(err, results) {
// ...
...format = function (sql,
values) {
if (typeof this.config.queryFormat === 'function') {
return this.config.queryFormat.call(this, sql, values, this.config.timezone);
}
return SqlString.format(sql, values, this.config.stringifyObjects, this.config.timezone);
}...
### Preparing Queries
You can use mysql.format to prepare a query with multiple insertion points, utilizing the proper escaping for ids and values. A \
simple example of this follows:
```js
var sql = "SELECT * FROM ?? WHERE ?? = ?";
var inserts = ['users', 'id', userId];
sql = mysql.format(sql, inserts);
```
Following this you then have a valid, escaped query that you can then send to the database safely. This is useful if you are loo\
king to prepare the query before actually sending it to the database. As mysql.format is exposed from SqlString.format you also \
have the option (but are not required) to pass in stringifyObject and timezone, allowing you provide a custom means of turning o\
bjects into strings, as well as a location-specific/timezone-aware Date.
### Custom format
If you prefer to have another type of query escape format, there's a connection configuration option you can use to define \
a custom format function. You can access the connection object if you want to use the built-in `.escape()` or any othe\
r connection function.
...pause = function () {
this._socket.pause();
this._protocol.pause();
}...
// Handle error, an 'end' event will be emitted after this as well
})
.on('fields', function(fields) {
// the field packets for the rows to follow
})
.on('result', function(row) {
// Pausing the connnection is useful if your processing involves I/O
connection.pause();
processRow(row, function() {
connection.resume();
});
})
.on('end', function() {
// all rows have been received
...function ping(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
this._implyConnect();
this._protocol.ping(options, bindToCurrentDomain(callback));
}...
## Ping
A ping packet can be sent over a connection using the `connection.ping` method. This
method will send a ping packet to the server and when the server responds, the callback
will fire. If an error occurred, the callback will fire with an error argument.
```js
connection.ping(function (err) {
if (err) throw err;
console.log('Server responded to ping');
})
```
## Timeouts
...function query(sql,
values,
cb) {
var query = Connection.createQuery(sql, values, cb);
query._connection = this;
if (!(typeof sql === 'object' && 'typeCast' in sql)) {
query.typeCast = this.config.typeCast;
}
if (query.sql) {
query.sql = this.format(query.sql, query.values);
}
this._implyConnect();
return this._protocol._enqueue(query);
}...
user : 'me',
password : 'secret',
database : 'my_db'
});
connection.connect();
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
connection.end();
```
...resume = function () {
this._socket.resume();
this._protocol.resume();
}...
// the field packets for the rows to follow
})
.on('result', function(row) {
// Pausing the connnection is useful if your processing involves I/O
connection.pause();
processRow(row, function() {
connection.resume();
});
})
.on('end', function() {
// all rows have been received
});
```
...function rollback(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
options = options || {};
options.sql = 'ROLLBACK';
options.values = null;
return this.query(options, callback);
}...
Simple transaction support is available at the connection level:
```js
connection.beginTransaction(function(err) {
if (err) { throw err; }
connection.query('INSERT INTO posts SET title=?', title, function(err, result) {
if (err) {
return connection.rollback(function() {
throw err;
});
}
var log = 'Post ' + result.insertId + ' added';
connection.query('INSERT INTO log SET data=?', log, function(err, result) {
...function statistics(options,
callback) {
if (!callback && typeof options === 'function') {
callback = options;
options = {};
}
this._implyConnect();
this._protocol.stats(options, bindToCurrentDomain(callback));
}n/a
function Pool(options) {
EventEmitter.call(this);
this.config = options.config;
this.config.connectionConfig.pool = this;
this._acquiringConnections = [];
this._allConnections = [];
this._freeConnections = [];
this._connectionQueue = [];
this._closed = false;
}n/a
function EventEmitter() {
EventEmitter.init.call(this);
}n/a
function acquireConnection(connection,
cb) {
if (connection._pool !== this) {
throw new Error('Connection acquired from wrong pool.');
}
var changeUser = this._needsChangeUser(connection);
var pool = this;
this._acquiringConnections.push(connection);
function onOperationComplete(err) {
spliceConnection(pool._acquiringConnections, connection);
if (pool._closed) {
err = new Error('Pool is closed.');
err.code = 'POOL_CLOSED';
}
if (err) {
pool._connectionQueue.unshift(cb);
pool._purgeConnection(connection);
return;
}
if (changeUser) {
pool.emit('connection', connection);
}
cb(null, connection);
}
if (changeUser) {
// restore user back to pool configuration
connection.config = this.config.newConnectionConfig();
connection.changeUser({timeout: this.config.acquireTimeout}, onOperationComplete);
} else {
// ping connection
connection.ping({timeout: this.config.acquireTimeout}, onOperationComplete);
}
}n/a
end = function (cb) {
this._closed = true;
if (typeof cb !== 'function') {
cb = function (err) {
if (err) throw err;
};
}
var calledBack = false;
var waitingClose = 0;
function onEnd(err) {
if (!calledBack && (err || --waitingClose <= 0)) {
calledBack = true;
cb(err);
}
}
while (this._allConnections.length !== 0) {
waitingClose++;
this._purgeConnection(this._allConnections[0], onEnd);
}
if (waitingClose === 0) {
process.nextTick(onEnd);
}
}...
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
connection.end();
```
From this example, you can learn the following:
* Every method you invoke on a connection is queued and executed in sequence.
* Closing the connection is done using `end()` which makes sure all remaining
queries are executed before sending a quit packet to the mysql server.
...escape = function (value) {
return mysql.escape(value, this.config.connectionConfig.stringifyObjects, this.config.connectionConfig.timezone);
}...
);
```
## Escaping query values
In order to avoid SQL Injection attacks, you should always escape any user
provided data before using it inside a SQL query. You can do so using the
`mysql.escape()`, `connection.escape()` or `pool.escape()&#x\
60; methods:
```js
var userId = 'some user provided value';
var sql = 'SELECT * FROM users WHERE id = ' + connection.escape(userId);
connection.query(sql, function(err, results) {
// ...
});
...function escapeId(value) {
return mysql.escapeId(value, false);
}...
console.log(query); // SELECT * FROM posts WHERE title='Hello MySQL'
```
## Escaping query identifiers
If you can't trust an SQL identifier (database / table / column name) because it is
provided by a user, you should escape it with `mysql.escapeId(identifier)`,
`connection.escapeId(identifier)` or `pool.escapeId(identifier)` like this:
```js
var sorter = 'date';
var sql = 'SELECT * FROM posts ORDER BY ' + connection.escapeId(sorter);
connection.query(sql, function(err, results) {
// ...
...getConnection = function (cb) {
if (this._closed) {
var err = new Error('Pool is closed.');
err.code = 'POOL_CLOSED';
process.nextTick(function () {
cb(err);
});
return;
}
var connection;
var pool = this;
if (this._freeConnections.length > 0) {
connection = this._freeConnections.shift();
this.acquireConnection(connection, cb);
return;
}
if (this.config.connectionLimit === 0 || this._allConnections.length < this.config.connectionLimit) {
connection = new PoolConnection(this, { config: this.config.newConnectionConfig() });
this._acquiringConnections.push(connection);
this._allConnections.push(connection);
connection.connect({timeout: this.config.acquireTimeout}, function onConnect(err) {
spliceConnection(pool._acquiringConnections, connection);
if (pool._closed) {
err = new Error('Pool is closed.');
err.code = 'POOL_CLOSED';
}
if (err) {
pool._purgeConnection(connection);
cb(err);
return;
}
pool.emit('connection', connection);
cb(null, connection);
});
return;
}
if (!this.config.waitForConnections) {
process.nextTick(function(){
var err = new Error('No connections available.');
err.code = 'POOL_CONNLIMIT';
cb(err);
});
return;
}
this._enqueueCallback(cb);
}...
var pool = mysql.createPool({
host : 'example.org',
user : 'bob',
password : 'secret',
database : 'my_db'
});
pool.getConnection(function(err, connection) {
// connected! (unless `err` is set)
});
```
When you are done with a connection, just call `connection.release()` and the
connection will return to the pool, ready to be used again by someone else.
...query = function (sql,
values,
cb) {
var query = Connection.createQuery(sql, values, cb);
if (!(typeof sql === 'object' && 'typeCast' in sql)) {
query.typeCast = this.config.connectionConfig.typeCast;
}
if (this.config.connectionConfig.trace) {
// Long stack trace support
query._callSite = new Error();
}
this.getConnection(function (err, conn) {
if (err) {
query.on('error', function () {});
query.end(err);
return;
}
// Release connection based off event
query.once('end', function() {
conn.release();
});
conn.query(query);
});
return query;
}...
user : 'me',
password : 'secret',
database : 'my_db'
});
connection.connect();
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
connection.end();
```
...function releaseConnection(connection) {
var pool = this;
if (this._acquiringConnections.indexOf(connection) !== -1) {
// connection is being acquired
return;
}
if (connection._pool) {
if (connection._pool !== this) {
throw new Error('Connection released to wrong pool');
}
if (this._freeConnections.indexOf(connection) !== -1) {
// connection already in free connection pool
// this won't catch all double-release cases
throw new Error('Connection already released');
} else {
// add connection to end of free queue
this._freeConnections.push(connection);
}
}
if (this._closed) {
// empty the connection queue
this._connectionQueue.splice(0).forEach(function (cb) {
var err = new Error('Pool is closed.');
err.code = 'POOL_CLOSED';
process.nextTick(function () {
cb(err);
});
});
} else if (this._connectionQueue.length) {
// get connection with next waiting callback
this.getConnection(this._connectionQueue.shift());
}
}n/a
function PoolCluster(config) {
EventEmitter.call(this);
config = config || {};
this._canRetry = typeof config.canRetry === 'undefined' ? true : config.canRetry;
this._defaultSelector = config.defaultSelector || 'RR';
this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
this._restoreNodeTimeout = config.restoreNodeTimeout || 0;
this._closed = false;
this._findCaches = Object.create(null);
this._lastId = 0;
this._namespaces = Object.create(null);
this._nodes = Object.create(null);
}n/a
function EventEmitter() {
EventEmitter.init.call(this);
}n/a
function add(id,
config) {
if (this._closed) {
throw new Error('PoolCluster is closed.');
}
var nodeId = typeof id === 'object'
? 'CLUSTER::' + (++this._lastId)
: String(id);
if (this._nodes[nodeId] !== undefined) {
throw new Error('Node ID "' + nodeId + '" is already defined in PoolCluster.');
}
var poolConfig = typeof id !== 'object'
? new PoolConfig(config)
: new PoolConfig(id);
this._nodes[nodeId] = {
id : nodeId,
errorCount : 0,
pool : new Pool({config: poolConfig}),
_offlineUntil : 0
};
this._clearFindCaches();
}...
PoolCluster provides multiple hosts connection. (group & retry & selector)
```js
// create
var poolCluster = mysql.createPoolCluster();
// add configurations (the config is a pool config object)
poolCluster.add(config); // add configuration with automatic name
poolCluster.add('MASTER', masterConfig); // add a named configuration
poolCluster.add('SLAVE1', slave1Config);
poolCluster.add('SLAVE2', slave2Config);
// remove configurations
poolCluster.remove('SLAVE2'); // By nodeId
poolCluster.remove('SLAVE*'); // By target group : SLAVE1-2
...function end(callback) {
var cb = callback !== undefined
? callback
: _cb;
if (typeof cb !== 'function') {
throw TypeError('callback argument must be a function');
}
if (this._closed) {
process.nextTick(cb);
return;
}
this._closed = true;
var calledBack = false;
var nodeIds = Object.keys(this._nodes);
var waitingClose = 0;
function onEnd(err) {
if (!calledBack && (err || --waitingClose <= 0)) {
calledBack = true;
cb(err);
}
}
for (var i = 0; i < nodeIds.length; i++) {
var nodeId = nodeIds[i];
var node = this._nodes[nodeId];
waitingClose++;
node.pool.end(onEnd);
}
if (waitingClose === 0) {
process.nextTick(onEnd);
}
}...
connection.query('SELECT 1 + 1 AS solution', function(err, rows, fields) {
if (err) throw err;
console.log('The solution is: ', rows[0].solution);
});
connection.end();
```
From this example, you can learn the following:
* Every method you invoke on a connection is queued and executed in sequence.
* Closing the connection is done using `end()` which makes sure all remaining
queries are executed before sending a quit packet to the mysql server.
...getConnection = function (pattern,
selector,
cb) {
var namespace;
if (typeof pattern === 'function') {
cb = pattern;
namespace = this.of();
} else {
if (typeof selector === 'function') {
cb = selector;
selector = this._defaultSelector;
}
namespace = this.of(pattern, selector);
}
namespace.getConnection(cb);
}...
var pool = mysql.createPool({
host : 'example.org',
user : 'bob',
password : 'secret',
database : 'my_db'
});
pool.getConnection(function(err, connection) {
// connected! (unless `err` is set)
});
```
When you are done with a connection, just call `connection.release()` and the
connection will return to the pool, ready to be used again by someone else.
...of = function (pattern,
selector) {
pattern = pattern || '*';
selector = selector || this._defaultSelector;
selector = selector.toUpperCase();
if (typeof PoolSelector[selector] === 'undefined') {
selector = this._defaultSelector;
}
var key = pattern + selector;
if (typeof this._namespaces[key] === 'undefined') {
this._namespaces[key] = new PoolNamespace(this, pattern, selector);
}
return this._namespaces[key];
}...
poolCluster.on('remove', function (nodeId) {
console.log('REMOVED NODE : ' + nodeId); // nodeId = SLAVE1
});
poolCluster.getConnection('SLAVE*', 'ORDER', function (err, connection) {});
// of namespace : of(pattern, selector)
poolCluster.of('*').getConnection(function (err, connection) {});
var pool = poolCluster.of('SLAVE*', 'RANDOM');
pool.getConnection(function (err, connection) {});
pool.getConnection(function (err, connection) {});
pool.query(function (err, result) {});
// close all connections
...function remove(pattern) {
var foundNodeIds = this._findNodeIds(pattern, true);
for (var i = 0; i < foundNodeIds.length; i++) {
var node = this._getNode(foundNodeIds[i]);
if (node) {
this._removeNode(node);
}
}
}...
// add configurations (the config is a pool config object)
poolCluster.add(config); // add configuration with automatic name
poolCluster.add('MASTER', masterConfig); // add a named configuration
poolCluster.add('SLAVE1', slave1Config);
poolCluster.add('SLAVE2', slave2Config);
// remove configurations
poolCluster.remove('SLAVE2'); // By nodeId
poolCluster.remove('SLAVE*'); // By target group : SLAVE1-2
// Target Group : ALL(anonymous, MASTER, SLAVE1-2), Selector : round-robin(default)
poolCluster.getConnection(function (err, connection) {});
// Target Group : MASTER, Selector : round-robin
poolCluster.getConnection('MASTER', function (err, connection) {});
...function Query(options,
callback) {
Sequence.call(this, options, callback);
this.sql = options.sql;
this.values = options.values;
this.typeCast = (options.typeCast === undefined)
? true
: options.typeCast;
this.nestTables = options.nestTables || false;
this._resultSet = null;
this._results = [];
this._fields = [];
this._index = 0;
this._loadError = null;
}n/a
function Sequence(options,
callback) {
if (typeof options === 'function') {
callback = options;
options = {};
}
EventEmitter.call(this);
options = options || {};
this._callback = callback;
this._callSite = null;
this._ended = false;
this._timeout = options.timeout;
// For Timers
this._idleNext = null;
this._idlePrev = null;
this._idleStart = null;
this._idleTimeout = -1;
this._repeat = null;
}n/a
EofPacket = function (packet) {
this._resultSet.eofPackets.push(packet);
if (this._resultSet.eofPackets.length === 1 && !this._callback) {
this.emit('fields', this._resultSet.fieldPackets, this._index);
}
if (this._resultSet.eofPackets.length !== 2) {
return;
}
if (this._callback) {
this._results.push(this._resultSet.rows);
this._fields.push(this._resultSet.fieldPackets);
}
this._index++;
this._resultSet = null;
this._handleFinalResultPacket(packet);
}n/a
ErrorPacket = function (packet) {
var err = this._packetToError(packet);
var results = (this._results.length > 0)
? this._results
: undefined;
var fields = (this._fields.length > 0)
? this._fields
: undefined;
err.index = this._index;
this.end(err, results, fields);
}n/a
FieldPacket = function (packet) {
this._resultSet.fieldPackets.push(packet);
}n/a
OkPacket = function (packet) {
// try...finally for exception safety
try {
if (!this._callback) {
this.emit('result', packet, this._index);
} else {
this._results.push(packet);
this._fields.push(undefined);
}
} finally {
this._index++;
this._resultSet = null;
this._handleFinalResultPacket(packet);
}
}n/a
ResultSetHeaderPacket = function (packet) {
this._resultSet = new ResultSet(packet);
// used by LOAD DATA LOCAL INFILE queries
if (packet.fieldCount === null) {
this._sendLocalDataFile(packet.extra);
}
}n/a
RowDataPacket = function (packet,
parser,
connection) {
packet.parse(parser, this._resultSet.fieldPackets, this.typeCast, this.nestTables, connection);
if (this._callback) {
this._resultSet.rows.push(packet);
} else {
this.emit('result', packet, this._index);
}
}n/a
determinePacket = function (firstByte,
parser) {
if (firstByte === 0) {
// If we have a resultSet and got one eofPacket
if (this._resultSet && this._resultSet.eofPackets.length === 1) {
// Then this is a RowDataPacket with an empty string in the first column.
// See: https://github.com/mysqljs/mysql/issues/222
} else if (this._resultSet && this._resultSet.resultSetHeaderPacket
&& this._resultSet.resultSetHeaderPacket.fieldCount !== null) {
return Packets.FieldPacket;
} else {
return undefined;
}
}
if (firstByte === 255) {
return undefined;
}
// EofPacket's are 5 bytes in mysql >= 4.1
// This is the only / best way to differentiate their firstByte from a 9
// byte length coded binary.
if (firstByte === 0xfe && parser.packetLength() < 9) {
return Packets.EofPacket;
}
if (!this._resultSet) {
return Packets.ResultSetHeaderPacket;
}
return (this._resultSet.eofPackets.length === 0)
? Packets.FieldPacket
: Packets.RowDataPacket;
}n/a
start = function () {
this.emit('packet', new Packets.ComQueryPacket(this.sql));
}n/a
stream = function (options) {
var self = this,
stream;
options = options || {};
options.objectMode = true;
stream = new Readable(options);
stream._read = function() {
self._connection && self._connection.resume();
};
stream.once('end', function() {
process.nextTick(function () {
stream.emit('close');
});
});
this.on('result',function(row,i) {
if (!stream.push(row)) self._connection.pause();
stream.emit('result',row,i); // replicate old emitter
});
this.on('error',function(err) {
stream.emit('error',err); // Pass on any errors
});
this.on('end', function() {
stream.push(null); // pushing null, indicating EOF
});
this.on('fields',function(fields,i) {
stream.emit('fields',fields,i); // replicate old emitter
});
return stream;
}...
Additionally you may be interested to know that it is currently not possible to
stream individual row columns, they will always be buffered up entirely. If you
have a good use case for streaming large fields to and from MySQL, I'd love to
get your thoughts and contributions on this.
### Piping results with Streams2
The query object provides a convenience method `.stream([options])` that wr\
aps
query events into a [Readable](http://nodejs.org/api/stream.html#stream_class_stream_readable)
[Streams2](http://blog.nodejs.org/2012/12/20/streams2/) object. This
stream can easily be piped downstream and provides automatic pause/resume,
based on downstream congestion and the optional `highWaterMark`. The
`objectMode` parameter of the stream is set to `true` and cannot be changed
(if you need a byte stream, you will need to use a transform stream, like
[objstream](https://www.npmjs.com/package/objstream) for example).
...