mirror of
https://github.com/mariadb-corporation/mariadb-connector-nodejs.git
synced 2025-07-29 11:59:44 +00:00
[misc] Promise implementation for connect() and end() methods
This commit is contained in:
rusher
@ -8,18 +8,22 @@ TODO have automatically generated table of contents
|
||||
|
||||
common API to mysql/mysql2:
|
||||
|
||||
* `connect(callback)`: Connect event with callback
|
||||
* `changeUser(options, callback)`: change current connection user
|
||||
* `beginTransaction(options, callback)`: begin transaction
|
||||
* `commit(options, callback)`: commit current transaction if any
|
||||
* `rollback(options, callback)`: rollback current transaction if any
|
||||
* `ping(options, callback)`: send an empty packet to server to check that connection is active
|
||||
* `connect([callback]) => Promise`: connect to database. <br/>
|
||||
Callback parameter is for compatibility with existing drivers.<br/>
|
||||
return Promise when no callback
|
||||
* `changeUser([options][,callback])`: change current connection user
|
||||
* `beginTransaction([options][,callback])`: begin transaction
|
||||
* `commit([options][,callback])`: commit current transaction if any
|
||||
* `rollback([options][,callback])`: rollback current transaction if any
|
||||
* `ping([options][,callback])`: send an empty packet to server to check that connection is active
|
||||
* `query(sql[, values][,callback])`: execute a [query](#query).
|
||||
* `pause()`: pause socket output.
|
||||
* `resume()`: resume socket output.
|
||||
* `on(eventName, listener)`: register to connection event
|
||||
* `once(eventName, listener)`: register to next connection event
|
||||
* `end(callback)`: gracefully end connection
|
||||
* `end([callback]) => Promise`: gracefully end connection<br/>
|
||||
Callback parameter is for compatibility with existing drivers.<br/>
|
||||
return Promise when no callback
|
||||
* `destroy()`: force connection ending.
|
||||
|
||||
|
||||
|
||||
@ -50,49 +50,49 @@ function Connection(options) {
|
||||
* Connect event with callback.
|
||||
*
|
||||
* @param callback(error)
|
||||
*
|
||||
* @returns {Promise} promise if no callback
|
||||
*/
|
||||
this.connect = callback => {
|
||||
switch (_status) {
|
||||
case Status.NOT_CONNECTED:
|
||||
_status = Status.CONNECTING;
|
||||
if (callback) _onConnect = callback;
|
||||
if (callback) {
|
||||
_registerHandshakeCmd(callback, callback);
|
||||
break;
|
||||
}
|
||||
return new Promise(function(resolve, reject) {
|
||||
_registerHandshakeCmd(resolve, reject);
|
||||
});
|
||||
|
||||
_registerHandshakeCmd();
|
||||
_initSocket();
|
||||
break;
|
||||
|
||||
case Status.CLOSING:
|
||||
case Status.CLOSED:
|
||||
if (!callback) return;
|
||||
callback(
|
||||
Errors.createError(
|
||||
"Connection closed",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_CONNECTION_ALREADY_CLOSED
|
||||
)
|
||||
const err = Errors.createError(
|
||||
"Connection closed",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_CONNECTION_ALREADY_CLOSED
|
||||
);
|
||||
break;
|
||||
if (callback) return callback(err);
|
||||
return Promise.reject(err);
|
||||
|
||||
case Status.CONNECTING:
|
||||
case Status.AUTHENTICATING:
|
||||
if (!callback) return;
|
||||
callback(
|
||||
Errors.createError(
|
||||
"Connection is already connecting",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_ALREADY_CONNECTING
|
||||
)
|
||||
const errConnecting = Errors.createError(
|
||||
"Connection is already connecting",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_ALREADY_CONNECTING
|
||||
);
|
||||
break;
|
||||
if (callback) return callback(errConnecting);
|
||||
return Promise.reject(errConnecting);
|
||||
|
||||
case Status.CONNECTED:
|
||||
if (!callback) return;
|
||||
callback();
|
||||
if (callback) callback();
|
||||
return Promise.resolve();
|
||||
}
|
||||
};
|
||||
|
||||
@ -248,25 +248,41 @@ function Connection(options) {
|
||||
* Terminate connection gracefully.
|
||||
*
|
||||
* @param callback when done
|
||||
* @returns {*} quit command
|
||||
* @returns {Promise} promise when no callback
|
||||
*/
|
||||
this.end = callback => {
|
||||
_addCommand = _addCommandDisabled;
|
||||
if (_status === Status.CONNECTING || _status === Status.CONNECTED) {
|
||||
_status = Status.CLOSING;
|
||||
const cmd = new Quit(() => {
|
||||
let sock = _socket;
|
||||
_clear();
|
||||
_status = Status.CLOSED;
|
||||
if (callback) setImmediate(callback);
|
||||
sock.destroy();
|
||||
});
|
||||
_sendQueue.push(cmd);
|
||||
_receiveQueue.push(cmd);
|
||||
let quitCmd;
|
||||
let promise;
|
||||
if (callback) {
|
||||
quitCmd = new Quit(() => {
|
||||
let sock = _socket;
|
||||
_clear();
|
||||
_status = Status.CLOSED;
|
||||
setImmediate(callback);
|
||||
sock.destroy();
|
||||
});
|
||||
} else {
|
||||
promise = new Promise(function(resolve, reject) {
|
||||
quitCmd = new Quit(() => {
|
||||
let sock = _socket;
|
||||
_clear();
|
||||
_status = Status.CLOSED;
|
||||
setImmediate(resolve);
|
||||
sock.destroy();
|
||||
});
|
||||
});
|
||||
}
|
||||
_sendQueue.push(quitCmd);
|
||||
_receiveQueue.push(quitCmd);
|
||||
if (_sendQueue.length === 1) {
|
||||
process.nextTick(_nextSendCmd.bind(this));
|
||||
}
|
||||
return promise;
|
||||
}
|
||||
return Promise.resolve();
|
||||
};
|
||||
|
||||
/**
|
||||
@ -411,27 +427,17 @@ function Connection(options) {
|
||||
// internal methods
|
||||
//*****************************************************************
|
||||
|
||||
/**
|
||||
* Default method called when connection is established (socket + authentication)
|
||||
*
|
||||
* @param err error if any
|
||||
* @private
|
||||
*/
|
||||
const _defaultOnConnect = function(err) {
|
||||
if (err && this.listenerCount("error") === 0) {
|
||||
throw err;
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Add handshake command to queue.
|
||||
*
|
||||
* @private
|
||||
*/
|
||||
const _registerHandshakeCmd = () => {
|
||||
const _registerHandshakeCmd = (resolve, rejected) => {
|
||||
var authenticationHandler = _authenticationEnd.bind(this, resolve, rejected);
|
||||
|
||||
const handshake = new Handshake(
|
||||
_authenticationEnd.bind(this),
|
||||
_createSecureContext.bind(this),
|
||||
authenticationHandler,
|
||||
_createSecureContext.bind(this, rejected),
|
||||
_addCommand.bind(this),
|
||||
_getSocket
|
||||
);
|
||||
@ -442,6 +448,7 @@ function Connection(options) {
|
||||
});
|
||||
|
||||
_receiveQueue.push(handshake);
|
||||
_initSocket(authenticationHandler, rejected);
|
||||
};
|
||||
|
||||
const _getSocket = () => {
|
||||
@ -452,7 +459,7 @@ function Connection(options) {
|
||||
* Initialize socket and associate events.
|
||||
* @private
|
||||
*/
|
||||
const _initSocket = () => {
|
||||
const _initSocket = (authenticationHandler, rejected) => {
|
||||
if (opts.socketPath) {
|
||||
_socket = Net.connect(opts.socketPath);
|
||||
} else {
|
||||
@ -460,9 +467,14 @@ function Connection(options) {
|
||||
}
|
||||
|
||||
if (opts.connectTimeout) {
|
||||
_socket.setTimeout(opts.connectTimeout, _connectTimeoutReached.bind(this));
|
||||
_socket.setTimeout(
|
||||
opts.connectTimeout,
|
||||
_connectTimeoutReached.bind(this, authenticationHandler)
|
||||
);
|
||||
}
|
||||
|
||||
const _socketError = _socketErrorHandler.bind(this, rejected);
|
||||
|
||||
_socket.on("data", _in.onData.bind(_in));
|
||||
_socket.on("error", _socketError);
|
||||
_socket.on("end", _socketError);
|
||||
@ -470,7 +482,10 @@ function Connection(options) {
|
||||
_socket.on("connect", () => {
|
||||
_status = Status.AUTHENTICATING;
|
||||
_socketConnected = true;
|
||||
_socket.setTimeout(opts.socketTimeout, _socketTimeoutReached.bind(this));
|
||||
_socket.setTimeout(
|
||||
opts.socketTimeout,
|
||||
_socketTimeoutReached.bind(this, authenticationHandler)
|
||||
);
|
||||
_socket.setNoDelay(true);
|
||||
});
|
||||
|
||||
@ -484,9 +499,9 @@ function Connection(options) {
|
||||
*
|
||||
* @private
|
||||
*/
|
||||
const _authenticationEnd = err => {
|
||||
process.nextTick(_onConnect, err);
|
||||
const _authenticationEnd = (resolve, reject, err) => {
|
||||
if (err) {
|
||||
process.nextTick(reject, err);
|
||||
//remove handshake command
|
||||
_receiveQueue.shift();
|
||||
|
||||
@ -504,6 +519,7 @@ function Connection(options) {
|
||||
}
|
||||
|
||||
if (opts.pipelining) _addCommand = _addCommandEnablePipeline;
|
||||
process.nextTick(resolve);
|
||||
_status = Status.CONNECTED;
|
||||
}
|
||||
};
|
||||
@ -514,7 +530,7 @@ function Connection(options) {
|
||||
* @param callback callback function when done
|
||||
* @private
|
||||
*/
|
||||
const _createSecureContext = callback => {
|
||||
const _createSecureContext = (rejected, callback) => {
|
||||
if (!tls.connect) {
|
||||
_fatalError(
|
||||
Errors.createError(
|
||||
@ -527,6 +543,8 @@ function Connection(options) {
|
||||
);
|
||||
}
|
||||
|
||||
const _socketError = _socketErrorHandler.bind(this, rejected);
|
||||
|
||||
const sslOption = Object.assign({}, opts.ssl, {
|
||||
servername: opts.host,
|
||||
socket: _socket
|
||||
@ -637,9 +655,9 @@ function Connection(options) {
|
||||
*
|
||||
* @private
|
||||
*/
|
||||
const _connectTimeoutReached = function() {
|
||||
const _connectTimeoutReached = function(authenticationHandler) {
|
||||
const handshake = _receiveQueue.peek();
|
||||
_authenticationEnd(
|
||||
authenticationHandler(
|
||||
Errors.createError(
|
||||
"Connection timeout",
|
||||
true,
|
||||
@ -661,7 +679,6 @@ function Connection(options) {
|
||||
_socket.destroy && _socket.destroy();
|
||||
_receiveQueue.shift(); //remove handshake packet
|
||||
const err = Errors.createError("socket timeout", true, info, "08S01", Errors.ER_SOCKET_TIMEOUT);
|
||||
process.nextTick(_onConnect, err);
|
||||
_fatalError(err, true);
|
||||
};
|
||||
|
||||
@ -745,50 +762,48 @@ function Connection(options) {
|
||||
* @returns {Function} socket error handle
|
||||
* @private
|
||||
*/
|
||||
const _socketErrorHandler = function(self) {
|
||||
return function(err) {
|
||||
switch (_status) {
|
||||
case Status.AUTHENTICATING:
|
||||
_authenticationEnd(err);
|
||||
break;
|
||||
const _socketErrorHandler = function(reject, err) {
|
||||
switch (_status) {
|
||||
case Status.AUTHENTICATING:
|
||||
_authenticationEnd(null, reject, err);
|
||||
break;
|
||||
|
||||
case Status.CLOSING:
|
||||
case Status.CLOSED:
|
||||
//already handled
|
||||
break;
|
||||
case Status.CLOSING:
|
||||
case Status.CLOSED:
|
||||
//already handled
|
||||
break;
|
||||
|
||||
default:
|
||||
//avoid sending new data in closed socket
|
||||
_socket.writeBuf = () => {};
|
||||
_socket.flush = () => {};
|
||||
default:
|
||||
//avoid sending new data in closed socket
|
||||
_socket.writeBuf = () => {};
|
||||
_socket.flush = () => {};
|
||||
|
||||
//socket has been ended without error
|
||||
if (!err) {
|
||||
if (_socketConnected) {
|
||||
err = Errors.createError(
|
||||
"socket has unexpectedly been closed",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_SOCKET_UNEXPECTED_CLOSE
|
||||
);
|
||||
} else {
|
||||
err = Errors.createError(
|
||||
"socket connection failed to established",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_SOCKET_CREATION_FAIL
|
||||
);
|
||||
}
|
||||
//socket has been ended without error
|
||||
if (!err) {
|
||||
if (_socketConnected) {
|
||||
err = Errors.createError(
|
||||
"socket has unexpectedly been closed",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_SOCKET_UNEXPECTED_CLOSE
|
||||
);
|
||||
} else {
|
||||
err = Errors.createError(
|
||||
"socket connection failed to established",
|
||||
true,
|
||||
info,
|
||||
"08S01",
|
||||
Errors.ER_SOCKET_CREATION_FAIL
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
//socket fail between socket creation and before authentication
|
||||
if (_status === Status.CONNECTING) process.nextTick(_onConnect, err);
|
||||
//socket fail between socket creation and before authentication
|
||||
if (_status === Status.CONNECTING) process.nextTick(reject, err);
|
||||
|
||||
_fatalError(err, false);
|
||||
}
|
||||
};
|
||||
_fatalError(err, false);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
@ -801,6 +816,7 @@ function Connection(options) {
|
||||
const _fatalErrorHandler = function(self) {
|
||||
return function(err, avoidThrowError) {
|
||||
if (_status === Status.CLOSING || _status === Status.CLOSED) return;
|
||||
const mustThrowError = _status !== Status.CONNECTING;
|
||||
_status = Status.CLOSING;
|
||||
|
||||
//prevent executing new commands
|
||||
@ -824,16 +840,18 @@ function Connection(options) {
|
||||
process.nextTick(receiveCmd.throwError.bind(receiveCmd), err);
|
||||
}
|
||||
}
|
||||
|
||||
if (self.listenerCount("error") > 0) {
|
||||
self.emit("error", err);
|
||||
self.emit("end");
|
||||
_clear();
|
||||
} else {
|
||||
self.emit("end");
|
||||
_clear();
|
||||
//error will be thrown if no error listener and no command did throw the exception
|
||||
if (!avoidThrowError && !errorThrownByCmd) throw err;
|
||||
if (mustThrowError) {
|
||||
//TODO to be removed when all use promise
|
||||
if (self.listenerCount("error") > 0) {
|
||||
self.emit("error", err);
|
||||
self.emit("end");
|
||||
_clear();
|
||||
} else {
|
||||
self.emit("end");
|
||||
_clear();
|
||||
//error will be thrown if no error listener and no command did throw the exception
|
||||
if (!avoidThrowError && !errorThrownByCmd) throw err;
|
||||
}
|
||||
}
|
||||
};
|
||||
};
|
||||
@ -871,8 +889,6 @@ function Connection(options) {
|
||||
const _sendQueue = new Queue();
|
||||
const _receiveQueue = new Queue();
|
||||
const _fatalError = _fatalErrorHandler(this);
|
||||
const _socketError = _socketErrorHandler(this);
|
||||
let _onConnect = _defaultOnConnect.bind(this);
|
||||
let _status = Status.NOT_CONNECTED;
|
||||
let _socketConnected = false;
|
||||
let _socket = null;
|
||||
@ -880,10 +896,6 @@ function Connection(options) {
|
||||
let _out = new PacketOutputStream(opts, info);
|
||||
let _in = new PacketInputStream(_unexpectedPacket.bind(this), _receiveQueue, _out, opts, info);
|
||||
|
||||
this.once("connect", err => {
|
||||
process.nextTick(_onConnect, err);
|
||||
});
|
||||
|
||||
//add alias threadId for mysql/mysql2 compatibility
|
||||
Object.defineProperty(this, "threadId", {
|
||||
get() {
|
||||
|
||||
@ -5,7 +5,7 @@ const assert = require("chai").assert;
|
||||
const Collations = require("../../lib/const/collations.js");
|
||||
|
||||
describe("connection", () => {
|
||||
it("multiple connection.connect() call", function(done) {
|
||||
it("multiple connection.connect() with callback", function(done) {
|
||||
const conn = base.createConnection();
|
||||
conn.connect(err => {
|
||||
if (err) done(err);
|
||||
@ -23,6 +23,52 @@ describe("connection", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("multiple connection.connect() with promise", function(done) {
|
||||
const conn = base.createConnection();
|
||||
conn
|
||||
.connect()
|
||||
.then(() => {
|
||||
return conn.connect();
|
||||
})
|
||||
.then(() => {
|
||||
return conn.end();
|
||||
})
|
||||
.then(() => {
|
||||
return conn.end();
|
||||
})
|
||||
.then(() => {
|
||||
conn
|
||||
.connect()
|
||||
.then(() => {
|
||||
done(new Error("must have thrown error"));
|
||||
})
|
||||
.catch(err => {
|
||||
assert.isTrue(err.message.includes("Connection closed"));
|
||||
done();
|
||||
});
|
||||
})
|
||||
.catch(done);
|
||||
});
|
||||
|
||||
it("multiple simultaneous connection.connect()", function(done) {
|
||||
const conn = base.createConnection();
|
||||
conn.connect().then(() => {
|
||||
return conn.end();
|
||||
});
|
||||
conn
|
||||
.connect()
|
||||
.then(() => {
|
||||
done(new Error("must have thrown error"));
|
||||
})
|
||||
.catch(err => {
|
||||
assert.equal(
|
||||
err.message,
|
||||
"(conn=-1, no: 45002, SQLState: 08S01) Connection is already connecting"
|
||||
);
|
||||
done();
|
||||
});
|
||||
});
|
||||
|
||||
it("connection event subscription", function(done) {
|
||||
let eventNumber = 0;
|
||||
const conn = base.createConnection();
|
||||
@ -118,8 +164,7 @@ describe("connection", () => {
|
||||
it("connection timeout error (wrong url)", done => {
|
||||
const initTime = Date.now();
|
||||
const conn = base.createConnection({ host: "www.google.fr", connectTimeout: 1000 });
|
||||
conn.connect();
|
||||
conn.on("error", err => {
|
||||
conn.connect().catch(err => {
|
||||
assert.strictEqual(err.message, "(conn=-1, no: 45012, SQLState: 08S01) Connection timeout");
|
||||
assert.isTrue(
|
||||
Date.now() - initTime >= 999,
|
||||
@ -255,10 +300,9 @@ describe("connection", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("connection on error event", function(done) {
|
||||
it("connection on error promise", function(done) {
|
||||
const conn = base.createConnection({ user: "fooUser" });
|
||||
conn.connect();
|
||||
conn.on("error", err => {
|
||||
conn.connect().catch(err => {
|
||||
if (!err) {
|
||||
done(new Error("must have thrown error"));
|
||||
} else done();
|
||||
|
||||
Reference in New Issue
Block a user