Ensure pool.getConnection retrieves good connections

dougwilson · dougwilson · commit 5ea649d6d94f · 2014-04-27T14:17:31.000-04:00
fixes #434 fixes #557 fixes #778
diff --git a/Changes.md b/Changes.md
@@ -18,6 +18,7 @@ you spot any mistakes.
 * Fix `pool.releaseConnection` to keep connection queue flowing #797
 * Fix SSL handshake error to  be catchable #800
 * Add `connection.threadId` to get MySQL connection ID #602
+* Ensure `pool.getConnection` retrieves good connections #434 #557 #778
 
 ## v2.1.1 (2014-03-13)
 
diff --git a/Readme.md b/Readme.md
@@ -304,6 +304,9 @@ up to 100 connections, but only ever use 5 simultaneously, only 5 connections
 will be made. Connections are also cycled round-robin style, with connections
 being taken from the top of the pool and returning to the bottom.
 
+When a previous connection is retrieved from the pool, a ping packet is sent
+to the server to check if the connection is still good.
+
 ## Pool options
 
 Pools accept all the same options as a connection. When creating a new
diff --git a/lib/Pool.js b/lib/Pool.js
@@ -31,9 +31,7 @@ Pool.prototype.getConnection = function (cb) {
   if (this._freeConnections.length > 0) {
     connection = this._freeConnections.shift();
 
-    return process.nextTick(function(){
-      cb(null, connection);
-    });
+    return this.acquireConnection(connection, cb);
   }
 
   if (this.config.connectionLimit === 0 || this._allConnections.length < this.config.connectionLimit) {
@@ -69,6 +67,27 @@ Pool.prototype.getConnection = function (cb) {
   this._connectionQueue.push(cb);
 };
 
+Pool.prototype.acquireConnection = function acquireConnection(connection, cb) {
+  if (connection._pool !== this) {
+    throw new Error('Connection acquired from wrong pool.');
+  }
+
+  var pool = this;
+
+  connection._pool = null;
+  connection.ping(function(err){
+    if (!err) {
+      connection._pool = pool;
+      cb(null, connection);
+      return;
+    }
+
+    connection.destroy();
+    pool._connectionQueue.unshift(cb);
+    pool._removeConnection(connection);
+  });
+};
+
 Pool.prototype.releaseConnection = function releaseConnection(connection) {
   var cb;
 
diff --git a/test/FakeServer.js b/test/FakeServer.js
@@ -131,6 +131,10 @@ FakeConnection.prototype._parsePacket = function(header) {
     case Packets.ComQueryPacket:
       this.emit('query', packet);
       break;
+    case Packets.ComPingPacket:
+      this._sendPacket(new Packets.OkPacket());
+      this._parser.resetPacketNumber();
+      break;
     case Packets.ComQuitPacket:
       this.emit('quit', packet);
       this._socket.end();
@@ -151,6 +155,7 @@ FakeConnection.prototype._determinePacket = function() {
   switch (firstByte) {
     case 0x01: return Packets.ComQuitPacket;
     case 0x03: return Packets.ComQueryPacket;
+    case 0x0e: return Packets.ComPingPacket;
     default:
       throw new Error('Unknown packet, first byte: ' + firstByte);
       break;
diff --git a/test/unit/pool/test-connection-reset.js b/test/unit/pool/test-connection-reset.js
@@ -0,0 +1,49 @@
+var assert = require('assert');
+var common = require('../../common');
+var pool   = common.createPool({
+  connectionLimit : 1,
+  port            : common.fakeServerPort
+});
+var server = common.createFakeServer();
+
+var seq        = 0;
+var serverConn = null;
+var tid        = 0;
+
+server.listen(common.fakeServerPort, function(err){
+  assert.ifError(err);
+
+  pool.getConnection(function(err, conn){
+    assert.ifError(err);
+    assert.equal(conn.threadId, 1);
+    conn.release();
+
+    // server destroys connection in pool
+    // read ECONNRESET
+    serverConn.destroy();
+
+    pool.getConnection(function(err, conn){
+      assert.ifError(err);
+      assert.equal(++seq, 1);
+      assert.equal(conn.threadId, 2);
+      conn.ping(function(err){
+        assert.ifError(err);
+        conn.release();
+      });
+    });
+
+    pool.getConnection(function(err, conn){
+      assert.ifError(err);
+      assert.equal(++seq, 2);
+      assert.equal(conn.threadId, 2);
+      server.destroy();
+    });
+  });
+});
+
+server.on('connection', function(incomingConnection) {
+  serverConn = incomingConnection;
+  incomingConnection.handshake({
+    threadId: ++tid
+  });
+});
