use the connector-manager to query the database

lib/sequelize/connector-manager.js

+var Query   = require("./query")
+  , without = function(arr, elem) { return arr.filter(function(e) { return e != elem }) }
+
 var ConnectorManager = module.exports = function(config) {
   this.client = null
   this.config = config
-  this.clientRequests = 0
   this.disconnectTimeoutId = null
+  this.queue = []
+  this.activeQueue = []
+  this.maxConcurrentQueries = (this.config.maxConcurrentQueries || 50)
 }
 
 ConnectorManager.prototype.connect = function() {
@@ -16,44 +21,83 @@ ConnectorManager.prototype.connect = function() {
     database: this.config.database
   })
 
- this.client.setMaxListeners(999)
+ this.client.setMaxListeners(1000000)
+}
+
+ConnectorManager.prototype.query = function(sql, callee, options) {
+  if(!this.isConnected) this.connect()
+
+  var queueItem = {
+    query: new Query(this.client, callee, options || {}),
+    sql: sql
+  }
+
+  this._enqueue(queueItem)
+
+  return queueItem.query
 }
 
 ConnectorManager.prototype.disconnect = function() {
-  this.client.end()
+  this.client.end(function() {})
 }
 
 ConnectorManager.prototype.reconnect = function() {
-  this.clientRequests = 0
   this.disconnect()
   this.connect()
 }
 
-ConnectorManager.prototype.afterQuery = function() {
-  this.clientRequests++
+// private
+
+ConnectorManager.prototype._enqueue = function(queueItem) {
+  if(this.activeQueue.length < this.maxConcurrentQueries) {
+    this.activeQueue.push(queueItem)
+    this._execQueueItem(queueItem)
+  } else {
+    this.queue.push(queueItem)
+  }
+}
+
+ConnectorManager.prototype._dequeue = function(queueItem) {
+  this.activeQueue = without(this.activeQueue, queueItem)
+}
+
+ConnectorManager.prototype._transferQueuedItems = function(count) {
+  for(var i = 0; i < count; i++) {
+    var queueItem = this.queue[0]
+    if(queueItem) {
+      this._enqueue(queueItem)
+      this.queue = without(this.queue, queueItem)
+    }
+  }
+}
 
-  this._reconnectIfTooManyConnections()
+ConnectorManager.prototype._afterQuery = function(queueItem) {
+  var self = this
+
+  this._dequeue(queueItem)
+  this._transferQueuedItems(this.maxConcurrentQueries - this.activeQueue.length)
   this._disconnectIfNoConnections()
 }
 
-ConnectorManager.prototype.__defineGetter__('tooManyRequests', function() {
-  return (this.clientRequests > 50)
-})
+
+ConnectorManager.prototype._execQueueItem = function(queueItem) {
+  var self = this
+
+  queueItem.query
+    .on('success', function(){ self._afterQuery(queueItem) })
+    .on('failure', function(){ self._afterQuery(queueItem) })
+
+  queueItem.query.run(queueItem.sql)
+}
 
 ConnectorManager.prototype.__defineGetter__('hasNoConnections', function() {
-  return this.client._queue && (this.client._queue.length == 0)
+  return (this.queue.length == 0) && (this.activeQueue.length == 0) && this.client._queue && (this.client._queue.length == 0)
 })
 
 ConnectorManager.prototype.__defineGetter__('isConnected', function() {
   return this.client != null
 })
 
-// private
-
-ConnectorManager.prototype._reconnectIfTooManyConnections = function() {
-  this.tooManyRequests && this.reconnect()
-}
-
 ConnectorManager.prototype._disconnectIfNoConnections = function() {
   var self = this
   

lib/sequelize/sequelize.js

 var Utils           = require("./utils")
   , ModelDefinition = require("./model-definition")
-  , Query           = require("./query")
   , DataTypes       = require('./data-types')
 
 var Sequelize = module.exports = function(database, username, password, options) {
@@ -39,21 +38,13 @@ var instanceMethods = {
   },
   
   query: function(sql, callee, options) {
-    options = options || {}
-    
+    options = options || {}
+
     if(this.options.queryOptions) options = Sequelize.Utils.merge(options, this.options.queryOptions)
-    if(!this.connectorManager.isConnected) this.connectorManager.connect()
-    
+
     options.logging = this.options.hasOwnProperty('logging') ? this.options.logging : true
-    
-    var self  = this
-      , query = new Query(this.connectorManager.client, callee, options).run(sql)
-    
-    query
-      .on('success', function(){ self.connectorManager.afterQuery() })
-      .on('failure', function(){ self.connectorManager.afterQuery() })
-    
-    return query
+
+    return this.connectorManager.query(sql, callee, options)
   },
   
   sync: function(options) {
@@ -68,9 +59,9 @@ var instanceMethods = {
       self.modelManager.models.forEach(function(model) { chainer.add(model.sync(options)) })
       
       chainer
-      .run()
-      .on('success', function() { eventEmitter.emit('success', null) })
-      .on('failure', function(err) { eventEmitter.emit('failure', err) })
+        .run()
+        .on('success', function() { eventEmitter.emit('success', null) })
+        .on('failure', function(err) { eventEmitter.emit('failure', err) })
     })
     return eventEmitter.run()
   }

lib/sequelize/utils.js

 var client = new (require("mysql").Client)()
 var Utils = module.exports = {
   _: (function() {
-    var _ = require("underscore");
-    _.mixin(require('underscore.string'));
+    var _ = require("underscore")
+
+    _.mixin(require('underscore.string'))
     _.mixin({
       camelizeIf: function(string, condition) {
         var result = string
@@ -15,6 +16,7 @@ var Utils = module.exports = {
         return result
       }
     })
+
     return _
   })(),
   addEventEmitter: function(_class) {

test/Model/save.js

@@ -37,8 +37,8 @@ module.exports = {
             assert.eql(true, uNow.getTime() < u.updatedAt.getTime())
             exit(function(){})
           })
-        }, 100)
-      }, 100)
+        }, 10)
+      }, 10)
     })
   }
 }
\ No newline at end of file