Handle all cases

lib/dialects/mysql/connector-manager.js

@@ -152,21 +152,22 @@ module.exports = (function() {
   Utils._.extend(ConnectorManager.prototype, require("../connector-manager").prototype);
 
   ConnectorManager.prototype.query = function(sql, callee, options) {
-    if (!this.isConnected && !this.pool) {
-      this.connect()
-    }
-
     if (this.useQueue) {
+      // If queueing we'll let the execQueueItem method handle connecting
       var queueItem = {
-        query: new Query(this.client, this.sequelize, callee, options || {}),
+        query: new Query(null, this.sequelize, callee, options || {}),
         sql: sql
       };
 
       queueItem.query.options.uuid = this.config.uuid
-      enqueue.call(this, queueItem, options)
+      enqueue.call(this, queueItem, options);
       return queueItem.query;
     }
 
+    if (!this.isConnected && !this.pool) {
+      this.connect()
+    }
+
     var self = this, query = new Query(this.client, this.sequelize, callee, options || {});
     this.pendingQueries++;
 
@@ -178,30 +179,23 @@ module.exports = (function() {
       } else {
         if (self.pendingQueries === 0) {
           setTimeout(function() {
-            self.pendingQueries === 0 && self.disconnect.call(self)
+            if (self.pendingQueries === 0){
+              self.disconnect.call(self);
+            } 
           }, 100);
         }
       }
     });
 
-    if (!this.pool) {
-      query.run(sql);
-    } else {
-      this.pool.acquire(function(err, client) {
-        if (err) {
-          return query.emit('error', err)
-        }
-
-        query.client = client
-        query.run(sql)
-        return;
-      }, undefined, options.type)
-    }
+    this.getConnection(options, function (err, client) {
+      query.client = client
+      query.run(sql)
+    });
 
     return query;
   };
 
-  ConnectorManager.prototype.connect = function() {
+  ConnectorManager.prototype.connect = function(callback) {
     var self = this;
     // in case database is slow to connect, prevent orphaning the client
     if (this.isConnecting || this.pool) {
@@ -209,11 +203,57 @@ module.exports = (function() {
     }
     connect.call(self, function(err, client) {
       self.client = client;
+      callback(err, client);
       return;
     });
     return;
   };
 
+  ConnectorManager.prototype.getConnection = function(options, callback) {
+    var self = this;
+
+    if (typeof options === "function") {
+      callback = options;
+      options = {};
+    }
+
+    return new Utils.CustomEventEmitter(function (emitter) {
+      if (!self.pool) {
+        if (self.client) {
+          return emitter.emit('success', self.client);
+        } else {
+          // Cache for concurrency
+          if (self._getConnection) {
+            self._getConnection.proxy(emitter);
+            return;
+          }
+
+          // Set cache and acquire connection
+          self._getConnection = emitter;
+          connect.call(self, function(err, client) {
+            if (err) {
+              return emitter.emit('error', err);
+            }
+
+            // Unset caching, should now be caught by the self.client check
+            self._getConnection = null;
+            self.client = client;
+            emitter.emit('success', client);
+          });
+        }
+      }
+      if (self.pool) {
+        // Acquire from pool
+        self.pool.acquire(function(err, client) {
+          if (err) {
+            return emitter.emit('error', err);
+          }
+          emitter.emit('success', client);
+        }, options.priority, options.type);
+      }
+    }).run().done(callback);
+  };
+
   ConnectorManager.prototype.disconnect = function() {
     if (this.client) {
       disconnect.call(this, this.client)
@@ -326,23 +366,7 @@ module.exports = (function() {
     options = options || {}
     if (this.activeQueue.length < this.maxConcurrentQueries) {
       this.activeQueue.push(queueItem)
-      if (this.pool) {
-        var self = this
-        this.pool.acquire(function(err, client) {
-          if (err) {
-            queueItem.query.emit('error', err)
-            return
-          }
-          //we set the client here, asynchronously, when getting a pooled connection
-          //allowing the ConnectorManager.query method to remain synchronous
-          queueItem.query.client = client
-          queueItem.client = client
-          execQueueItem.call(self, queueItem)
-          return
-        }, undefined, options.type)
-      } else {
-        execQueueItem.call(this, queueItem)
-      }
+      execQueueItem.call(this, queueItem)
     } else {
       this.queue.push(queueItem)
     }
@@ -375,11 +399,23 @@ module.exports = (function() {
   var execQueueItem = function(queueItem) {
     var self = this
 
-    queueItem.query
-      .success(function(){ afterQuery.call(self, queueItem) })
-      .error(function(){ afterQuery.call(self, queueItem) })
+    self.getConnection({
+      priority: queueItem.query.options.priority,
+      type: queueItem.query.options.type
+    }, function (err, connection) {
+      if (err) {
+        queueItem.query.emit('error', err)
+        return
+      }
 
-    queueItem.query.run(queueItem.sql, queueItem.client)
+      queueItem.query.client = connection
+      queueItem.client = connection
+      queueItem.query
+        .success(function(){ afterQuery.call(self, queueItem) })
+        .error(function(){ afterQuery.call(self, queueItem) })
+
+      queueItem.query.run(queueItem.sql, queueItem.client)
+    })
   }
 
   ConnectorManager.prototype.__defineGetter__('hasQueuedItems', function() {