refactor(connections): first step in refactoring connections to a more generic/u…

…nderstandable codebase

changelog.md

@@ -25,6 +25,7 @@ Notice: All 1.7.x changes are present in 2.0.x aswell
 - `sequelize.showAllSchemas` now returns an array of schemas, instead of an array containinig an array of schemas
 - `sequelize.showAllSchemas` now returns an array of schemas, instead of an array containinig an array of schemas
 - `sequelize.transaction()` now returns a promise rather than a instance of Sequelize.Transaction
 - `sequelize.transaction()` now returns a promise rather than a instance of Sequelize.Transaction
 - `bulkCreate`, `bulkUpdate` and `bulkDestroy` (and aliases) now take both a `hooks` and an `individualHooks` option, `hooks` defines whether or not to run the main hooks, and `individualHooks` defines whether to run hooks for each instance affected.
 - `bulkCreate`, `bulkUpdate` and `bulkDestroy` (and aliases) now take both a `hooks` and an `individualHooks` option, `hooks` defines whether or not to run the main hooks, and `individualHooks` defines whether to run hooks for each instance affected.
+- It is no longer possible to disable pooling, disable pooling will just result in a 1/1 pool.
 
 
 # v2.0.0-dev11
 # v2.0.0-dev11
 ### Caution: This release contains many changes and is highly experimental
 ### Caution: This release contains many changes and is highly experimental

lib/connection-manager.js

+"use strict";
+
+var Pooling = require('generic-pool')
+  , Promise = require('./promise')
+  , _ = require('lodash')
+  , defaultPoolingConfig = {
+    max: 10,
+    min: 0,
+    idle: 10000,
+    handleDisconnects: false
+  }
+  , ConnectionManager;
+
+ConnectionManager = function(sequelize) {
+  var config = sequelize.config
+    , self = this;
+
+  this.config = config;
+  this.dialect = sequelize.dialect;
+
+  if (config.pool) {
+    config.pool = _.extend({}, config.pool); // Make sure we don't modify the existing config object (user might re-use it)
+    config.pool =_.defaults(config.pool, defaultPoolingConfig, {
+      validate: this.$validate.bind(this)
+    }) ;
+  } else {
+    // If the user has turned off pooling we provide a 0/1 pool for backwards compat
+    config.pool = _.defaults({
+      max: 1,
+      min: 0,
+    }, defaultPoolingConfig, {
+      validate: this.$validate.bind(this)
+    });
+  }
+
+  // Map old names
+  if (config.pool.maxIdleTime) config.pool.idle = config.pool.maxIdleTime;
+  if (config.pool.maxConnections) config.pool.max = config.pool.maxConnections;
+  if (config.pool.minConnections) config.pool.min = config.pool.minConnections;
+
+  if (config.replication) {
+    var reads = 0
+      , writes = 0;
+
+    // Make sure we don't modify the existing config object (user might re-use it)
+    config.replication.write = _.extend({}, config.replication.write);
+    config.replication.read = config.replication.read.map(function (read) {
+      return _.extend({}, read);
+    });
+
+    // Map main connection config
+    config.replication.write = _.defaults(config.replication.write, {
+      host: config.host,
+      port: config.port,
+      username: config.username,
+      password: config.password,
+      database: config.database
+    });
+
+    for (var i = 0; i < config.replication.read.length; i++) {
+      config.replication.read[i] = _.defaults(config.replication.read[i], {
+        host: this.config.host,
+        port: this.config.port,
+        username: this.config.username,
+        password: this.config.password,
+        database: this.config.database
+      });
+    }
+
+    // I'll make my own pool, with blackjack and hookers! (original credit goes to @janzeh)
+    this.pool = {
+      release: function(client) {
+        if (client.queryType === 'read') {
+          return self.pool.read.release(client);
+        } else {
+          return self.pool.write.release(client);
+        }
+      },
+      acquire: function(callback, priority, queryType) {
+        if (queryType === 'SELECT') {
+          self.pool.read.acquire(callback, priority);
+        } else {
+          self.pool.write.acquire(callback, priority);
+        }
+      },
+      drain: function() {
+        self.pool.read.drain();
+        self.pool.write.drain();
+      },
+      read: Pooling.Pool({
+        name: 'sequelize-connection-read',
+        create: function(done) {
+          if (reads >= config.replication.read.length) {
+            reads = 0;
+          }
+          // Simple round robin config
+          self.$connect(config.replication.read[reads++]).then(function (connection) {
+            connection.queryType = 'read';
+          }).nodeify(done);
+        },
+        destroy: function(connection) {
+          self.$disconnect(connection);
+        },
+        validate: config.pool.validate,
+        max: config.pool.max,
+        min: config.pool.min,
+        idleTimeoutMillis: config.pool.idle
+      }),
+      write: Pooling.Pool({
+        name: 'sequelize-connection-write',
+        create: function(done) {
+          self.$connect(config.replication.write).then(function (connection) {
+            connection.queryType = 'write';
+          }).nodeify(done);
+        },
+        destroy: function(connection) {
+          self.$disconnect(connection);
+        },
+        validate: config.pool.validate,
+        max: config.pool.max,
+        min: config.pool.min,
+        idleTimeoutMillis: config.pool.idle
+      })
+    };
+  } else {
+    this.pool = Pooling.Pool({
+      name: 'sequelize-connection',
+      create: function(done) {
+        self.$connect(config).nodeify(done);
+      },
+      destroy: function(connection) {
+        self.$disconnect(connection);
+      },
+      max: config.pool.max,
+      min: config.pool.min,
+      validate: config.pool.validate,
+      idleTimeoutMillis: config.pool.idle
+    });
+  }
+
+  this.onProcessExit = function() {
+    // Cleanup
+    self.pool.drain();
+    return;
+  }.bind(this);
+
+  process.on('exit', this.onProcessExit);
+};
+
+ConnectionManager.prototype.getConnection = function(options) {
+  var self = this;
+  options = options || {};
+  return new Promise(function (resolve, reject) {
+    self.pool.acquire(function(err, connection) {
+      if (err) return reject(err);
+      resolve(connection);
+    }, options.priority, options.type);
+  });
+};
+ConnectionManager.prototype.releaseConnection = function(connection) {
+  var self = this;
+  return new Promise(function (resolve, reject) {
+    self.pool.release(connection);
+    resolve();
+  });
+};
+
+ConnectionManager.prototype.$connect = function(config) {
+  return this.dialect.connectionManager.connect(config);
+};
+ConnectionManager.prototype.$disconnect = function(connection) {
+  return this.dialect.connectionManager.disconnect(connection);
+};
+
+ConnectionManager.prototype.$validate = function(connection) {
+  if (!this.dialect.connectionManager.validate) return Promise.resolve();
+  return this.dialect.connectionManager.validate(connection);
+};
+
+module.exports = ConnectionManager;
\ No newline at end of file

lib/dialects/abstract/index.js

@@ -13,4 +13,4 @@ AbstractDialect.prototype.supports = {
   schemas: false
   schemas: false
 };
 };
 
 
-module.exports = AbstractDialect;
+module.exports = AbstractDialect;
\ No newline at end of file

lib/dialects/mariadb/connection-manager.js

+"use strict";
+var ConnectionManager
+  , Promise = require('../../promise');
+
+ConnectionManager = function(dialect, sequelize) {
+  try {
+    this.lib = require(sequelize.config.dialectModulePath || 'mariasql');
+  } catch (err) {
+    throw new Error('Please install mariasql package manually');
+  }
+};
+
+ConnectionManager.prototype.connect = function(config) {
+  return new Promise(function (resolve, reject) {
+
+  });
+};
+ConnectionManager.prototype.disconnect = function(connection) {
+  return new Promise(function (resolve, reject) {
+
+  });
+};
+
+module.exports = ConnectionManager;
\ No newline at end of file

lib/dialects/mariadb/index.js

 'use strict';
 'use strict';
 
 
 var _ = require('lodash')
 var _ = require('lodash')
-  , MySQL = require('../mysql');
+  , MySQL = require('../mysql')
+  , ConnectionManager = require('./connection-manager')
+  , Query = require('./query');
 
 
 var MariaDialect = function(sequelize) {
 var MariaDialect = function(sequelize) {
   this.sequelize = sequelize;
   this.sequelize = sequelize;
+  this.connectionManager = new ConnectionManager(this, sequelize);
 };
 };
 
 
 MariaDialect.prototype = _.defaults({
 MariaDialect.prototype = _.defaults({
-    'LIMIT ON UPDATE': true
+	'LIMIT ON UPDATE': true
 }, MySQL.prototype);
 }, MySQL.prototype);
 
 
-module.exports = MariaDialect;
+MariaDialect.prototype.Query = Query;
+
+module.exports = MariaDialect;
\ No newline at end of file

lib/dialects/mysql/connection-manager.js

+"use strict";
+var ConnectionManager
+  , Promise = require('../../promise');
+
+ConnectionManager = function(dialect, sequelize) {
+  try {
+    this.lib = require(sequelize.config.dialectModulePath || 'mysql');
+  } catch (err) {
+    throw new Error('Please install mysql package manually');
+  }
+};
+
+ConnectionManager.prototype.connect = function(config) {
+  var self = this;
+  return new Promise(function (resolve, reject) {
+    var connectionConfig = {
+      host: config.host,
+      port: config.port,
+      user: config.username,
+      password: config.password,
+      database: config.database,
+      timezone: 'Z'
+    };
+
+    if (config.dialectOptions) {
+      Object.keys(config.dialectOptions).forEach(function(key) {
+        connectionConfig[key] = config.dialectOptions[key];
+      });
+    }
+
+    var connection = self.lib.createConnection(connectionConfig);
+
+    connection.connect(function(err) {
+      if (err) {
+        switch (err.code) {
+        case 'ECONNREFUSED':
+        case 'ER_ACCESS_D2ENIED_ERROR':
+          reject('Failed to authenticate for MySQL. Please double check your settings.');
+          break;
+        case 'ENOTFOUND':
+        case 'EHOSTUNREACH':
+        case 'EINVAL':
+          reject('Failed to find MySQL server. Please double check your settings.');
+          break;
+        default:
+          reject(err);
+          break;
+        }
+
+        return;
+      }
+
+      resolve(connection);
+    });
+
+  }).tap(function (connection) {
+    connection.query("SET time_zone = '+0:00'");
+  });
+};
+ConnectionManager.prototype.disconnect = function(connection) {
+  return new Promise(function (resolve, reject) {
+    connection.end(function(err) {
+      if (err) return reject(err);
+      resolve();
+    });
+  });
+};
+ConnectionManager.prototype.validate = function(connection) {
+  return connection && connection.state !== 'disconnected';
+};
+
+module.exports = ConnectionManager;
\ No newline at end of file

lib/dialects/mysql/index.js

 'use strict';
 'use strict';
 
 
 var _ = require('lodash')
 var _ = require('lodash')
-  , Abstract = require('../abstract');
+  , Abstract = require('../abstract')
+  , ConnectionManager = require('./connection-manager')
+  , Query = require('./query');
 
 
 var MysqlDialect = function(sequelize) {
 var MysqlDialect = function(sequelize) {
   this.sequelize = sequelize;
   this.sequelize = sequelize;
+  this.connectionManager = new ConnectionManager(this, sequelize);
 };
 };
 
 
 MysqlDialect.prototype.supports = _.defaults({
 MysqlDialect.prototype.supports = _.defaults({
@@ -14,4 +17,6 @@ MysqlDialect.prototype.supports = _.defaults({
   forShare: 'LOCK IN SHARE MODE'
   forShare: 'LOCK IN SHARE MODE'
 }, Abstract.prototype.supports);
 }, Abstract.prototype.supports);
 
 
-module.exports = MysqlDialect;
+MysqlDialect.prototype.Query = Query;
+
+module.exports = MysqlDialect;
\ No newline at end of file

lib/dialects/mysql/query.js

@@ -5,8 +5,8 @@ var Utils = require('../../utils')
   , uuid = require('node-uuid');
   , uuid = require('node-uuid');
 
 
 module.exports = (function() {
 module.exports = (function() {
-  var Query = function(client, sequelize, callee, options) {
-    this.client = client;
+  var Query = function(connection, sequelize, callee, options) {
+    this.connection = connection;
     this.callee = callee;
     this.callee = callee;
     this.sequelize = sequelize;
     this.sequelize = sequelize;
     this.uuid = uuid.v4();
     this.uuid = uuid.v4();
@@ -18,10 +18,6 @@ module.exports = (function() {
 
 
     var self = this;
     var self = this;
     this.checkLoggingOption();
     this.checkLoggingOption();
-    this.promise = new Utils.Promise(function(resolve, reject) {
-      self.resolve = resolve;
-      self.reject = reject;
-    });
   };
   };
 
 
   Utils.inherit(Query, AbstractQuery);
   Utils.inherit(Query, AbstractQuery);
@@ -30,22 +26,24 @@ module.exports = (function() {
     this.sql = sql;
     this.sql = sql;
 
 
     if (this.options.logging !== false) {
     if (this.options.logging !== false) {
-      this.sequelize.log('Executing (' + this.options.uuid + '): ' + this.sql);
+      this.sequelize.log('Executing (' + this.connnection.uuid + '): ' + this.sql);
     }
     }
 
 
-    self.client.query(self.sql, function(err, results, fields) {
-      self.promise.emit('sql', self.sql, self.options.uuid);
+    var promise = new Utils.Promise(function(resolve, reject) {
+      self.connection.query(self.sql, function(err, results, fields) {
+        promise.emit('sql', self.sql, self.connection.uuid);
 
 
-      if (err) {
-        err.sql = sql;
+        if (err) {
+          err.sql = sql;
+          reject(err);
+        } else {
+          resolve(self.formatResults(results));
+        }
+      }).setMaxListeners(100);
 
 
-        self.reject(err);
-      } else {
-        self.resolve(self.formatResults(results));
-      }
-    }).setMaxListeners(100);
+    });
 
 
-    return this.promise;
+    return promise;
   };
   };
 
 
   return Query;
   return Query;

lib/dialects/postgres/connection-manager.js

+"use strict";
+var ConnectionManager
+  , Promise = require('../../promise');
+
+ConnectionManager = function(dialect, sequelize) {
+  this.sequelize = sequelize;
+  try {
+    this.lib = sequelize.config.native ? require(sequelize.config.dialectModulePath || 'pg').native : require(sequelize.config.dialectModulePath || 'pg');
+  } catch (err) {
+    throw new Error('Please install postgres package manually');
+  }
+};
+
+ConnectionManager.prototype.connect = function(config) {
+  var self = this;
+  return new Promise(function (resolve, reject) {
+    var connectionString = self.sequelize.getQueryInterface().QueryGenerator.databaseConnectionUri(config)
+      , connection = new self.lib.Client(connectionString)
+      , responded = false;
+
+    connection.connect(function(err) {
+      if (err) {
+        if (err.code) {
+          switch (err.code) {
+          case 'ECONNREFUSED':
+            reject(new Error('Failed to authenticate for PostgresSQL. Please double check your settings.'));
+            break;
+          case 'ENOTFOUND':
+          case 'EHOSTUNREACH':
+          case 'EINVAL':
+            reject(new Error('Failed to find PostgresSQL server. Please double check your settings.'));
+            break;
+          default:
+            reject(err);
+            break;
+          }
+        } else {
+          reject(new Error(err.message));
+        }
+        return;
+      }
+      responded = true;
+      resolve(connection);
+    });
+
+    // If we didn't ever hear from the client.connect() callback the connection timeout, node-postgres does not treat this as an error since no active query was ever emitted
+    connection.on('end', function () {
+      if (!responded) {
+        reject(new Error('Connection timed out'));
+      }
+    });
+  }).tap(function (connection) {
+    if (self.sequelize.config.keepDefaultTimezone) return;
+    return new Promise(function (resolve, reject) {
+      connection.query("SET TIME ZONE 'UTC'").on('error', function (err) {
+        reject(err);
+      }).on('end', function () {
+        resolve();
+      });
+    });
+  });
+};
+ConnectionManager.prototype.disconnect = function(connection) {
+  return new Promise(function (resolve, reject) {
+    connection.end();
+    resolve();
+  });
+};
+
+module.exports = ConnectionManager;
\ No newline at end of file

lib/dialects/postgres/index.js

 'use strict';
 'use strict';
 
 
 var _ = require('lodash')
 var _ = require('lodash')
-  , Abstract = require('../abstract');
+  , Abstract = require('../abstract')
+  , ConnectionManager = require('./connection-manager')
+  , Query = require('./query');
 
 
 var PostgresDialect = function(sequelize) {
 var PostgresDialect = function(sequelize) {
   this.sequelize = sequelize;
   this.sequelize = sequelize;
+  this.connectionManager = new ConnectionManager(this, sequelize);
 };
 };
 
 
 PostgresDialect.prototype.supports = _.defaults({
 PostgresDialect.prototype.supports = _.defaults({
@@ -15,4 +18,6 @@ PostgresDialect.prototype.supports = _.defaults({
   forShare: 'FOR SHARE'
   forShare: 'FOR SHARE'
 }, Abstract.prototype.supports);
 }, Abstract.prototype.supports);
 
 
-module.exports = PostgresDialect;
+PostgresDialect.prototype.Query = Query;
+
+module.exports = PostgresDialect;
\ No newline at end of file

lib/dialects/sqlite/connection-manager.js

+"use strict";
+var ConnectionManager
+  , Promise = require('../../promise');
+
+ConnectionManager = function(dialect, sequelize) {
+  this.sequelize = sequelize;
+
+  try {
+    this.lib = require(sequelize.config.dialectModulePath || 'sqlite3').verbose();
+  } catch (err) {
+    throw new Error('Please install mariasql package manually');
+  }
+};
+
+ConnectionManager.prototype.connect = function(config) {
+  var self = this;
+  return new Promise(function (resolve, reject) {
+    var connection = new self.lib.Database(self.sequelize.options.storage || ':memory:', function(err) {
+      if (err) {
+        if (err.code === 'SQLITE_CANTOPEN') return reject('Failed to find SQL server. Please double check your settings.');
+        return reject(err);
+      }
+      resolve(connection);
+    });
+  }).tap(function (connection) {
+    if (self.sequelize.options.foreignKeys !== false) {
+      // Make it possible to define and use foreign key constraints unless
+      // explicitly disallowed. It's still opt-in per relation
+      connection.run('PRAGMA FOREIGN_KEYS=ON');
+    }
+  });
+};
+ConnectionManager.prototype.disconnect = function(connection) {
+  return new Promise(function (resolve, reject) {
+    connection.close(function (err) {
+      if (err) return reject(err);
+      resolve();
+    });
+  });
+};
+
+module.exports = ConnectionManager;
\ No newline at end of file

lib/dialects/sqlite/index.js

 'use strict';
 'use strict';
 
 
 var _ = require('lodash')
 var _ = require('lodash')
-  , Abstract = require('../abstract');
+  , Abstract = require('../abstract')
+  , ConnectionManager = require('./connection-manager')
+  , Query = require('./query');
 
 
 var SqliteDialect = function(sequelize) {
 var SqliteDialect = function(sequelize) {
   this.sequelize = sequelize;
   this.sequelize = sequelize;
+  this.connectionManager = new ConnectionManager(this, sequelize);
 };
 };
 
 
 SqliteDialect.prototype.supports = _.defaults({
 SqliteDialect.prototype.supports = _.defaults({
@@ -12,4 +15,6 @@ SqliteDialect.prototype.supports = _.defaults({
   'DEFAULT VALUES': true
   'DEFAULT VALUES': true
 }, Abstract.prototype.supports);
 }, Abstract.prototype.supports);
 
 
-module.exports = SqliteDialect;
+SqliteDialect.prototype.Query = Query;
+
+module.exports = SqliteDialect;
\ No newline at end of file

lib/sequelize.js

@@ -7,8 +7,8 @@ var url = require('url')
   , DataTypes = require('./data-types')
   , DataTypes = require('./data-types')
   , ModelManager = require('./model-manager')
   , ModelManager = require('./model-manager')
   , QueryInterface = require('./query-interface')
   , QueryInterface = require('./query-interface')
+  , ConnectionManager = require('./connection-manager')
   , Transaction = require('./transaction')
   , Transaction = require('./transaction')
-  , TransactionManager = require('./transaction-manager')
   , QueryTypes = require('./query-types')
   , QueryTypes = require('./query-types')
   , sequelizeErrors = require('./errors')
   , sequelizeErrors = require('./errors')
   , Promise = require('./promise');
   , Promise = require('./promise');
@@ -161,7 +161,7 @@ module.exports = (function() {
       var Dialect = require('./dialects/' + this.getDialect());
       var Dialect = require('./dialects/' + this.getDialect());
       this.dialect = new Dialect(this);
       this.dialect = new Dialect(this);
     } catch (err) {
     } catch (err) {
-      throw new Error('The dialect ' + this.getDialect() + ' is not supported.');
+      throw new Error('The dialect ' + this.getDialect() + ' is not supported. ('+err+')');
     }
     }
 
 
     /**
     /**
@@ -170,7 +170,7 @@ module.exports = (function() {
      */
      */
     this.models = {};
     this.models = {};
     this.modelManager = this.daoFactoryManager = new ModelManager(this);
     this.modelManager = this.daoFactoryManager = new ModelManager(this);
-    this.transactionManager = new TransactionManager(this);
+    this.connectionManager = new ConnectionManager(this);
 
 
     this.importCache = {};
     this.importCache = {};
   };
   };
@@ -466,6 +466,8 @@ module.exports = (function() {
    * @see {Model#build} for more information about callee.
    * @see {Model#build} for more information about callee.
    */
    */
   Sequelize.prototype.query = function(sql, callee, options, replacements) {
   Sequelize.prototype.query = function(sql, callee, options, replacements) {
+    var self = this;
+
     if (arguments.length === 4) {
     if (arguments.length === 4) {
       if (Array.isArray(replacements)) {
       if (Array.isArray(replacements)) {
         sql = Utils.format([sql].concat(replacements), this.options.dialect);
         sql = Utils.format([sql].concat(replacements), this.options.dialect);
@@ -487,7 +489,15 @@ module.exports = (function() {
       type: (sql.toLowerCase().indexOf('select') === 0) ? QueryTypes.SELECT : false
       type: (sql.toLowerCase().indexOf('select') === 0) ? QueryTypes.SELECT : false
     });
     });
 
 
-    return this.transactionManager.query(sql, callee, options);
+    return Promise.resolve(
+      options.transaction ? options.transaction.connection : self.connectionManager.getConnection()
+    ).then(function (connection) {
+      var query = new self.dialect.Query(connection, self, callee, options);
+      return query.run(sql).finally(function() {
+        if (options.transaction) return;
+        return self.connectionManager.releaseConnection(connection);
+      });
+    });
   };
   };
 
 
   /**
   /**

lib/transaction-manager.js

-'use strict';
-
-var Utils = require('./utils');
-
-var TransactionManager = module.exports = function(sequelize) {
-  this.sequelize = sequelize;
-  this.connectorManagers = {};
-
-  try {
-    this.ConnectorManager = require('./dialects/' + sequelize.getDialect() + '/connector-manager');
-  } catch (err) {
-    throw new Error('The dialect ' + sequelize.getDialect() + ' is not supported.');
-  }
-};
-
-TransactionManager.prototype.getConnectorManager = function(uuid) {
-  uuid = uuid || 'default';
-
-  if (!this.connectorManagers.hasOwnProperty(uuid)) {
-    var config = Utils._.extend({ uuid: uuid }, this.sequelize.config);
-
-    if (uuid !== 'default') {
-      config.pool = Utils._.extend(
-        {},
-        Utils._.clone(config.pool || {}),
-        {
-          minConnections: 1,
-          maxConnections: 1,
-          useReplicaton: false
-        }
-      );
-      config.keepDefaultTimezone = true;
-    }
-
-    this.connectorManagers[uuid] = new this.ConnectorManager(this.sequelize, config);
-  }
-
-  return this.connectorManagers[uuid];
-};
-
-TransactionManager.prototype.releaseConnectionManager = function(uuid) {
-  this.connectorManagers[uuid].cleanup();
-  delete this.connectorManagers[uuid];
-};
-
-TransactionManager.prototype.query = function(sql, callee, options) {
-  options = options || {};
-  options.uuid = 'default';
-
-  if (options.transaction) {
-    options.uuid = options.transaction.id;
-  }
-
-  return this.getConnectorManager(options.uuid).query(sql, callee, options);
-};

lib/transaction.js

@@ -87,37 +87,34 @@ Transaction.prototype.rollback = function() {
 };
 };
 
 
 Transaction.prototype.prepareEnvironment = function() {
 Transaction.prototype.prepareEnvironment = function() {
-  var self = this
-    , connectorManager = self.sequelize.transactionManager.getConnectorManager(this.id);
+  var self = this;
 
 
-  return this.begin().then(function () {
+  return this.sequelize.connectionManager.getConnection().then(function (connection) {
+    self.connection = connection;
+    self.connection.uuid = self.id;
+  }).then(function () {
+    return self.begin();
+  }).then(function () {
     return self.setIsolationLevel();
     return self.setIsolationLevel();
   }).then(function () {
   }).then(function () {
     return self.setAutocommit();
     return self.setAutocommit();
-  }).then(function () {
-    return new Utils.Promise(function (resolve, reject) {
-      connectorManager.afterTransactionSetup(function (err, result) {
-        if (err) return reject(err);
-        return resolve(result);
-      });
-    });
   });
   });
 };
 };
-Transaction.prototype.begin = function(callback) {
+Transaction.prototype.begin = function() {
   return this
   return this
     .sequelize
     .sequelize
     .getQueryInterface()
     .getQueryInterface()
     .startTransaction(this, {});
     .startTransaction(this, {});
 };
 };
 
 
-Transaction.prototype.setAutocommit = function(callback) {
+Transaction.prototype.setAutocommit = function() {
   return this
   return this
     .sequelize
     .sequelize
     .getQueryInterface()
     .getQueryInterface()
     .setAutocommit(this, this.options.autocommit);
     .setAutocommit(this, this.options.autocommit);
 };
 };
 
 
-Transaction.prototype.setIsolationLevel = function(callback) {
+Transaction.prototype.setIsolationLevel = function() {
   return this
   return this
     .sequelize
     .sequelize
     .getQueryInterface()
     .getQueryInterface()
@@ -125,5 +122,5 @@ Transaction.prototype.setIsolationLevel = function(callback) {
 };
 };
 
 
 Transaction.prototype.cleanup = function() {
 Transaction.prototype.cleanup = function() {
-  this.sequelize.transactionManager.releaseConnectionManager(this.id);
+  return this.sequelize.connectionManager.releaseConnection(this.connection);
 };
 };

test/transaction-manager.test.js

-var chai               = require('chai')
-  , expect             = chai.expect
-  , Support            = require(__dirname + '/support')
-  , TransactionManager = require(__dirname + '/../lib/transaction-manager')
-
-describe(Support.getTestDialectTeaser("TransactionManager"), function () {
-  beforeEach(function() {
-    this.transactionManager = new TransactionManager(this.sequelize)
-  })
-
-  describe('getConnectorManager', function() {
-    describe('if no uuid is passed', function() {
-      it('uses the default connector', function() {
-        var connectorManager = this.transactionManager.getConnectorManager()
-        expect(connectorManager.config.uuid).to.equal('default')
-      })
-
-      it('uses the pooling configuration of sequelize', function() {
-        var connectorManager = this.transactionManager.getConnectorManager()
-          , self             = this
-
-        if (Support.getTestDialect() !== 'sqlite') {
-          expect(this.sequelize.config.pool.maxConnections).to.equal(5)
-        }
-
-        Object.keys(this.sequelize.config.pool || {}).forEach(function(key) {
-          expect(connectorManager.config.pool[key]).to.equal(self.sequelize.config.pool[key])
-        })
-      })
-    })
-
-    describe('if the passed uuid is not equal to "default"', function() {
-      it('uses the non-default connector', function() {
-        var connectorManager = this.transactionManager.getConnectorManager('a-uuid')
-        expect(connectorManager.config.uuid).to.equal('a-uuid')
-      })
-
-      it('creates a new connector manager', function() {
-        this.transactionManager.getConnectorManager()
-        expect(Object.keys(this.transactionManager.connectorManagers).length).to.equal(1)
-        this.transactionManager.getConnectorManager('a-uuid')
-        expect(Object.keys(this.transactionManager.connectorManagers).length).to.equal(2)
-        this.transactionManager.getConnectorManager('a-uuid')
-        expect(Object.keys(this.transactionManager.connectorManagers).length).to.equal(2)
-      })
-
-      it('treats the connector managers as singleton', function() {
-        var connectorManager1 = this.transactionManager.getConnectorManager('a-uuid')
-          , connectorManager2 = this.transactionManager.getConnectorManager('a-uuid')
-
-        expect(connectorManager1).to.equal(connectorManager2)
-      })
-
-      it('uses the pooling configuration of sequelize but with disabled replication and forced to one connection', function() {
-        var connectorManager = this.transactionManager.getConnectorManager('a-uuid')
-          , self             = this
-
-        if (Support.getTestDialect() !== 'sqlite') {
-          expect(this.sequelize.config.pool.maxConnections).to.equal(5)
-        }
-
-        Object.keys(this.sequelize.config.pool || {}).forEach(function(key) {
-          if (['minConnections', 'maxConnections'].indexOf(key) === -1) {
-            expect(connectorManager.config.pool[key]).to.equal(self.sequelize.config.pool[key])
-          }
-        })
-
-        expect(connectorManager.config.pool.minConnections).to.equal(1)
-        expect(connectorManager.config.pool.maxConnections).to.equal(1)
-      })
-    })
-  })
-})