Skip to content

Added isValid connection health check #264

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
he-is-harry wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Added isValid connection health check #264

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions lib/Client.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,11 @@ Client.prototype.setClientInfo = function setClientInfo(key, val) {
}
};

Client.prototype.isValid = function isValid(timeout, cb) {
// Timeout is inputted in seconds, convert to milliseconds
this._connection.isValid(timeout * 1000, cb);
}

Client.prototype._execute = function _execute(command, options, cb) {
var result = this._createResult(this._connection, options);
this._connection.executeDirect({
Expand Down
82 changes: 60 additions & 22 deletions lib/protocol/Connection.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,13 @@ Connection.prototype._addListeners = function _addListeners(socket) {
socket.removeListener('close', onclose);
}

function clearStateTimeout() {
if (self._state.timeoutObject) {
clearTimeout(self._state.timeoutObject);
self._state.timeoutObject = undefined;
}
}

// register listerners on socket
function ondata(chunk) {
packet.push(chunk);
Expand All @@ -258,17 +265,24 @@ Connection.prototype._addListeners = function _addListeners(socket) {
self._state.sessionId = packet.header.sessionId;
self._state.packetCount = -1;
}
var buffer = packet.getData();
packet.clear();
var cb = self._state.receive;
self._state.receive = undefined;
self._state.messageType = undefined;
self.receive(buffer, cb);
// Ensure the packet corresponds to the current request
if (self._state.packetCount === -1 || self._state.packetCount === packet.header.packetCount) {
var buffer = packet.getData();
packet.clear();
var cb = self._state.receive;
self._state.receive = undefined;
self._state.messageType = undefined;
clearStateTimeout();
self.receive(buffer, cb);
} else {
packet.clear();
}
}
}
socket.on('data', ondata);

function onerror(err) {
clearStateTimeout();
var cb = self._state && self._state.receive;
if (cb) {
self._state.receive = null; // a callback should be called only once
Expand Down Expand Up @@ -313,7 +327,7 @@ Connection.prototype._clearQueue = function _clearQueue(err) {
}
};

Connection.prototype.send = function send(message, receive) {
Connection.prototype.send = function send(message, options, receive) {
if (this._statementContext) {
message.unshift(PartKind.STATEMENT_CONTEXT, this._statementContext.getOptions());
}
Expand Down Expand Up @@ -354,6 +368,14 @@ Connection.prototype.send = function send(message, receive) {
if (this._socket) {
this._socket.write(packet);
}
var self = this;
function onTimeout() {
self._state.receive = undefined;
receive(new Error('Socket receive timeout (receive took longer than ' + options.communicationTimeout + ' ms)'));
}
if (options.communicationTimeout) {
state.timeoutObject = setTimeout(onTimeout, options.communicationTimeout);
}
};


Expand Down Expand Up @@ -422,7 +444,7 @@ Connection.prototype.receive = function receive(buffer, cb) {
}
};

Connection.prototype.enqueue = function enqueue(task, cb) {
Connection.prototype.enqueue = function enqueue(task, options, cb) {
var queueable;

if (!this._socket || !this._queue || this.readyState === 'closed') {
Expand All @@ -439,7 +461,7 @@ Connection.prototype.enqueue = function enqueue(task, cb) {
queueable.name = task.name;
} else if (util.isObject(task)) {
if (task instanceof request.Segment) {
queueable = this._queue.createTask(this.send.bind(this, task), cb);
queueable = this._queue.createTask(this.send.bind(this, task, options), cb);
queueable.name = MessageTypeName[task.type];
} else if (util.isFunction(task.run)) {
queueable = task;
Expand Down Expand Up @@ -574,7 +596,7 @@ Connection.prototype.connect = function connect(options, cb) {
clientId: self.clientId,
connectOptions: self.connectOptions.getOptions(),
useCesu8: self.useCesu8
}), connReceive);
}), options, connReceive);
});
}

Expand All @@ -588,7 +610,7 @@ Connection.prototype.connect = function connect(options, cb) {
authOptions.dbConnectInfo = true;
}

this.send(request.authenticate(authOptions), authReceive);
this.send(request.authenticate(authOptions), options, authReceive);
};

Connection.prototype.disconnect = function disconnect(cb) {
Expand All @@ -600,7 +622,7 @@ Connection.prototype.disconnect = function disconnect(cb) {
}

function enqueueDisconnect() {
self.enqueue(request.disconnect(), done);
self.enqueue(request.disconnect(), {}, done);
}

if (this.isIdle()) {
Expand All @@ -616,7 +638,7 @@ Connection.prototype.executeDirect = function executeDirect(options, cb) {
scrollableCursor: this.scrollableCursor,
useCesu8: this.useCesu8
}, options);
this.enqueue(request.executeDirect(options), cb);
this.enqueue(request.executeDirect(options), options, cb);
};

Connection.prototype.prepare = function prepare(options, cb) {
Expand All @@ -625,7 +647,7 @@ Connection.prototype.prepare = function prepare(options, cb) {
scrollableCursor: this.scrollableCursor,
useCesu8: this.useCesu8
}, options);
this.enqueue(request.prepare(options), cb);
this.enqueue(request.prepare(options), options, cb);
};

Connection.prototype.readLob = function readLob(options, cb) {
Expand All @@ -635,7 +657,7 @@ Connection.prototype.readLob = function readLob(options, cb) {
};
}
options.autoCommit = this.autoCommit;
this.enqueue(request.readLob(options), cb);
this.enqueue(request.readLob(options), options, cb);
};

Connection.prototype.execute = function execute(options, cb) {
Expand All @@ -646,40 +668,40 @@ Connection.prototype.execute = function execute(options, cb) {
parameters: EMPTY_BUFFER
}, options);
if (options.parameters === EMPTY_BUFFER) {
return this.enqueue(request.execute(options), cb);
return this.enqueue(request.execute(options), options, cb);
}
this.enqueue(createExecuteTask(this, options, cb));
};

Connection.prototype.fetchNext = function fetchNext(options, cb) {
options.autoCommit = this.autoCommit;
options.useCesu8 = this.useCesu8;
this.enqueue(request.fetchNext(options), cb);
this.enqueue(request.fetchNext(options), options, cb);
};

Connection.prototype.closeResultSet = function closeResultSet(options, cb) {
this.enqueue(request.closeResultSet(options), cb);
this.enqueue(request.closeResultSet(options), options, cb);
};

Connection.prototype.dropStatement = function dropStatement(options, cb) {
options.useCesu8 = this.useCesu8;
this.enqueue(request.dropStatementId(options), cb);
this.enqueue(request.dropStatementId(options), options, cb);
};

Connection.prototype.commit = function commit(options, cb) {
if (util.isFunction(options)) {
cb = options;
options = {};
}
this.enqueue(request.commit(options), cb);
this.enqueue(request.commit(options), options, cb);
};

Connection.prototype.rollback = function rollback(options, cb) {
if (util.isFunction(options)) {
cb = options;
options = {};
}
this.enqueue(request.rollback(options), cb);
this.enqueue(request.rollback(options), options, cb);
};

// The function doesn't use the queue. It's used before the queue starts running
Expand All @@ -689,7 +711,7 @@ Connection.prototype.fetchDbConnectInfo = function (options, cb) {
err.code = 'EHDBCLOSE';
return cb(err)
}
this.send(request.dbConnectInfo(options), function(err, reply) {
this.send(request.dbConnectInfo(options), options, function(err, reply) {
if (err) {
return cb(err);
}
Expand Down Expand Up @@ -728,6 +750,21 @@ Connection.prototype.destroy = function destroy(err) {
}
};

Connection.prototype.isValid = function isValid(timeout, cb) {
var options = {
command: 'SELECT 1 FROM DUMMY WHERE 1 = 0',
communicationTimeout: timeout > 0 ? timeout : undefined,
}
this.executeDirect(options, function (err, reply) {
if (err) {
// Currently, isValid will swallow all errors and indicate invalid
cb(null, false);
} else {
cb(null, true);
}
});
}

Connection.prototype.isIdle = function isIdle() {
return this._queue.empty && !this._queue.busy;
};
Expand Down Expand Up @@ -759,6 +796,7 @@ function ConnectionState() {
this.packetCount = -1;
this.messageType = undefined;
this.receive = undefined;
this.timeoutObject = undefined;
}

function Version(major, minor) {
Expand Down
20 changes: 12 additions & 8 deletions lib/protocol/ExecuteTask.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,14 +203,15 @@ ExecuteTask.prototype.sendExecute = function sendExecute(cb) {
if (err) {
return cb(err);
}
self.connection.send(request.execute({
var options = {
autoCommit: self.autoCommit,
holdCursorsOverCommit: self.holdCursorsOverCommit,
scrollableCursor: self.scrollableCursor,
statementId: self.statementId,
parameters: parameters,
useCesu8: self.connection.useCesu8
}), cb);
};
self.connection.send(request.execute(options), options, cb);
});
};

Expand All @@ -221,24 +222,27 @@ ExecuteTask.prototype.sendWriteLobRequest = function sendWriteLobRequest(cb) {
if (err) {
return cb(err);
}
self.connection.send(request.writeLob({
var options = {
writeLobRequest: buffer
}), cb);
};
self.connection.send(request.writeLob(options), options, cb);
});
};

ExecuteTask.prototype.sendCommit = function sendCommit(cb) {
var self = this;
self.connection.send(request.commit({
var options = {
holdCursorsOverCommit: self.holdCursorsOverCommit
}), cb);
};
self.connection.send(request.commit(options), options, cb);
};

ExecuteTask.prototype.sendRollback = function sendRollback(cb) {
var self = this;
self.connection.send(request.rollback({
var options = {
holdCursorsOverCommit: self.holdCursorsOverCommit
}), cb);
};
self.connection.send(request.rollback(options), options, cb);
};

function createInvalidFunctionCodeError() {
Expand Down
50 changes: 50 additions & 0 deletions test/acceptance/db.Events.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
'use strict';

var db = require('../db')();
var RemoteDB = require('../db/RemoteDB');
var describeRemoteDB = db instanceof RemoteDB ? describe : describe.skip;

describe('db', function () {

Expand Down Expand Up @@ -59,4 +61,52 @@ describe('db', function () {
});
});

describeRemoteDB('isValid', function () {
before(db.init.bind(db));
after(function (done) {
if (client.readyState !== 'closed') {
db.end(done);
} else {
done();
}
});
var client = db.client;

it('should be valid when connected', function (done) {
client.isValid(0, function (err, ret) { // no timeout
if (err) done(err);
ret.should.be.true();
done();
})
});

it('should be valid with timeout when connected', function (done) {
client.isValid(1, function (err, ret) { // 1 second timeout
if (err) done(err);
ret.should.be.true();
done();
});
});

it('should be invalid when disconnected', function (done) {
client.exec('SELECT CURRENT_CONNECTION FROM DUMMY', function (err, res) {
var connId = res[0].CURRENT_CONNECTION;
var adminDB = require('../db')();
adminDB.init(function (err) {
if (err) done(err);
var adminClient = adminDB.client;
var disconnectSQL = "ALTER SYSTEM DISCONNECT SESSION '" + connId + "'";
adminClient.exec(disconnectSQL, function (err) {
if (err) done(err);
client.isValid(0, function (err, ret) { // disconnected
if (err) done(err);
ret.should.be.false();
adminDB.end(done);
});
});
});
});
});
});

});
Loading