Add "query" method to PoolNamespace

closes #1256
closes #1505
closes #1506

Author: alsotang

Changes.md

@@ -6,6 +6,8 @@ you spot any mistakes.
 
 ## HEAD
 
+* Add `query` method to `PoolNamespace` #1256 #1505 #1506
+  - Used as `cluster.of(...).query(...)`
 * Update `bignumber.js` to 2.4.0
 * Update `sqlstring` to 2.1.0
   - Accept numbers and other value types in `escapeId`

Readme.md

@@ -468,6 +468,7 @@ 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
 poolCluster.end(function (err) {
@@ -538,7 +539,7 @@ space for a new connection to be created on the next getConnection call.
 ## Performing queries
 
 The most basic way to perform a query is to call the `.query()` method on an object
-(like a `Connection` or `Pool` instance).
+(like a `Connection`, `Pool`, or `PoolNamespace` instance).
 
 The simplest form of .`query()` is `.query(sqlString, callback)`, where a SQL string
 is the first argument and the second is a callback:

lib/PoolNamespace.js

@@ -1,3 +1,4 @@
+var Connection   = require('./Connection');
 var PoolSelector = require('./PoolSelector');
 
 module.exports = PoolNamespace;
@@ -54,6 +55,65 @@ PoolNamespace.prototype.getConnection = function(cb) {
   });
 };
 
+PoolNamespace.prototype.query = function (sql, values, cb) {
+  var cluster     = this._cluster;
+  var clusterNode = this._getClusterNode();
+  var query       = Connection.createQuery(sql, values, cb);
+  var namespace   = this;
+
+  if (clusterNode === null) {
+    var err = null;
+
+    if (this._cluster._findNodeIds(this._pattern, true).length !== 0) {
+      err = new Error('Pool does not have online node.');
+      err.code = 'POOL_NONEONLINE';
+    } else {
+      err = new Error('Pool does not exist.');
+      err.code = 'POOL_NOEXIST';
+    }
+
+    process.nextTick(function () {
+      query.on('error', function () {});
+      query.end(err);
+    });
+    return query;
+  }
+
+  if (!(typeof sql === 'object' && 'typeCast' in sql)) {
+    query.typeCast = clusterNode.pool.config.connectionConfig.typeCast;
+  }
+
+  if (clusterNode.pool.config.connectionConfig.trace) {
+    // Long stack trace support
+    query._callSite = new Error;
+  }
+
+  cluster._getConnection(clusterNode, function (err, conn) {
+    var retry = err && cluster._canRetry
+      && cluster._findNodeIds(namespace._pattern).length !== 0;
+
+    if (retry) {
+      namespace.query(query);
+      return;
+    }
+
+    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;
+};
+
 PoolNamespace.prototype._getClusterNode = function _getClusterNode() {
   var foundNodeIds = this._cluster._findNodeIds(this._pattern);
   var nodeId;

test/unit/pool-cluster/test-query-connection-error.js

@@ -0,0 +1,25 @@
+var assert = require('assert');
+var common = require('../../common');
+
+var cluster = common.createPoolCluster();
+var server  = common.createFakeServer();
+
+var poolConfig = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('SLAVE1', poolConfig);
+// cluster.add('SLAVE2', poolConfig);
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  var pool = cluster.of('SLAVE*', 'ORDER');
+
+  pool.query('SELECT 1', function (err, rows) {
+    assert.ok(err, 'got error');
+    assert.equal(err.code, 'ER_HOST_NOT_PRIVILEGED');
+    server.destroy();
+  });
+});
+
+server.on('connection', function (conn) {
+  conn.deny('You suck.', common.Errors.ER_HOST_NOT_PRIVILEGED);
+});

test/unit/pool-cluster/test-query-pool-error.js

@@ -0,0 +1,19 @@
+var assert = require('assert');
+var common = require('../../common');
+
+var cluster = common.createPoolCluster();
+var server  = common.createFakeServer();
+
+var poolConfig = common.getTestConfig({port: common.fakeServerPort});
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  var pool = cluster.of('SLAVE*', 'ORDER');
+
+  pool.query('SELECT 1', function (err, rows) {
+    assert.ok(err, 'got error');
+    assert.equal(err.code, 'POOL_NOEXIST');
+    server.destroy();
+  });
+});

test/unit/pool-cluster/test-query-restore.js

@@ -0,0 +1,53 @@
+var assert  = require('assert');
+var common  = require('../../common');
+var cluster = common.createPoolCluster({
+  canRetry             : true,
+  removeNodeErrorCount : 2,
+  restoreNodeTimeout   : 100
+});
+var server  = common.createFakeServer();
+
+var connCount  = 0;
+var offline    = true;
+var poolConfig = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('MASTER', poolConfig);
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  var pool = cluster.of('MASTER', 'ORDER');
+
+  pool.query('SELECT 1', function (err) {
+    assert.ok(err);
+    assert.equal(err.code, 'PROTOCOL_CONNECTION_LOST');
+    assert.equal(err.fatal, true);
+    assert.equal(connCount, 2);
+
+    pool.query('SELECT 1', function (err) {
+      assert.ok(err);
+      assert.equal(err.code, 'POOL_NONEONLINE');
+
+      offline = false;
+    });
+
+    setTimeout(function () {
+      pool.query('SELECT 1', function (err) {
+        assert.ifError(err);
+        cluster.end(function (err) {
+          assert.ifError(err);
+          server.destroy();
+        });
+      });
+    }, 200);
+  });
+});
+
+server.on('connection', function (conn) {
+  connCount += 1;
+
+  if (offline) {
+    conn.destroy();
+  } else {
+    conn.handshake();
+  }
+});

test/unit/pool-cluster/test-query.js

@@ -0,0 +1,25 @@
+var assert  = require('assert');
+var common  = require('../../common');
+var cluster = common.createPoolCluster();
+var server  = common.createFakeServer();
+
+var poolConfig = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('SLAVE1', poolConfig);
+cluster.add('SLAVE2', poolConfig);
+
+server.listen(common.fakeServerPort, function(err) {
+  assert.ifError(err);
+
+  var pool = cluster.of('SLAVE*', 'ORDER');
+
+  pool.query('SELECT 1', function (err, rows) {
+    assert.ifError(err);
+    assert.equal(rows.length, 1);
+    assert.equal(rows[0]['1'], 1);
+
+    cluster.end(function (err) {
+      assert.ifError(err);
+      server.destroy();
+    });
+  });
+});