Skip to content

public / sequelize

Go to a project

不要怂，就是干，撸起袖子干！

Commit 54269128 by Sushant Committed by GitHub
Options

refactor(abstract/connection-manager) (#8330) 

Resolves issue with acquire call which never finishes, resolve errors and connections and handle factory level error ourselves. Fixed issue with MSSQL / MySQL error out with resources currently not part of pool.

Fixed evict setting so idle connections are properly removed on time and improved documentation around connection manager and various options
1 parent e978e59b
Inline Side-by-side
Showing with 207 additions and 125 deletions
docker-compose.yml
...@@ -6,6 +6,7 @@ services: ...@@ -6,6 +6,7 @@ services:
links: links:
- mysql-57 - mysql-57
- postgres-95 - postgres-95
- mssql
volumes: volumes:
- .:/sequelize - .:/sequelize
environment: environment:
...@@ -39,3 +40,13 @@ services: ...@@ -39,3 +40,13 @@ services:
ports: ports:
- "127.0.0.1:8999:3306" - "127.0.0.1:8999:3306"
container_name: mysql-57 container_name: mysql-57
# MSSQL
mssql:
image: microsoft/mssql-server-linux:latest
environment:
ACCEPT_EULA: "Y"
SA_PASSWORD: yourStrong(!)Password
ports:
- "127.0.0.1:8997:1433"
container_name: mssql
lib/dialects/abstract/connection-manager.js
...@@ -6,17 +6,24 @@ const _ = require('lodash'); ...@@ -6,17 +6,24 @@ const _ = require('lodash');
const Utils = require('../../utils'); const Utils = require('../../utils');
const debug = Utils.getLogger().debugContext('pool'); const debug = Utils.getLogger().debugContext('pool');
const semver = require('semver'); const semver = require('semver');
const timers = require('timers');
const defaultPoolingConfig = { const defaultPoolingConfig = {
max: 5, max: 5,
min: 0, min: 0,
idle: 10000, idle: 10000,
acquire: 10000, acquire: 10000,
evict: 60000, evict: 10000,
handleDisconnects: true handleDisconnects: true
}; };
/**
* Abstract Connection Manager
*
* Connection manager which handles pool, replication and determining database version
* Works with generic-pool to maintain connection pool
*
* @private
*/
class ConnectionManager { class ConnectionManager {
constructor(dialect, sequelize) { constructor(dialect, sequelize) {
const config = _.cloneDeep(sequelize.config); const config = _.cloneDeep(sequelize.config);
...@@ -25,22 +32,20 @@ class ConnectionManager { ...@@ -25,22 +32,20 @@ class ConnectionManager {
this.config = config; this.config = config;
this.dialect = dialect; this.dialect = dialect;
this.versionPromise = null; this.versionPromise = null;
this.poolError = null;
this.dialectName = this.sequelize.options.dialect; this.dialectName = this.sequelize.options.dialect;
if (config.pool === false) { if (config.pool === false) {
throw new Error('Support for pool:false was removed in v4.0'); throw new Error('Support for pool:false was removed in v4.0');
} }
config.pool =_.defaults(config.pool || {}, defaultPoolingConfig, { config.pool = _.defaults(config.pool || {}, defaultPoolingConfig, {
validate: this._validate.bind(this), validate: this._validate.bind(this),
Promise Promise
}) ; });
// Save a reference to the bound version so we can remove it with removeListener // Save a reference to the bound version so we can remove it with removeListener
this.onProcessExit = this.onProcessExit.bind(this); this._onProcessExit = this._onProcessExit.bind(this);
process.on('exit', this._onProcessExit);
process.on('exit', this.onProcessExit);
this.initPools(); this.initPools();
} }
...@@ -57,7 +62,13 @@ class ConnectionManager { ...@@ -57,7 +62,13 @@ class ConnectionManager {
}); });
} }
onProcessExit() { /**
* Handler which executes on process exit or connection manager shutdown
*
* @private
* @return {Promise}
*/
_onProcessExit() {
if (!this.pool) { if (!this.pool) {
return Promise.resolve(); return Promise.resolve();
} }
...@@ -68,59 +79,55 @@ class ConnectionManager { ...@@ -68,59 +79,55 @@ class ConnectionManager {
}); });
} }
/**
* Drain the pool and close it permanently
*
* @return {Promise}
*/
close() { close() {
// Remove the listener, so all references to this instance can be garbage collected. // Remove the listener, so all references to this instance can be garbage collected.
process.removeListener('exit', this.onProcessExit); process.removeListener('exit', this._onProcessExit);
// Mark close of pool // Mark close of pool
this.getConnection = function getConnection() { this.getConnection = function getConnection() {
return Promise.reject(new Error('ConnectionManager.getConnection was called after the connection manager was closed!')); return Promise.reject(new Error('ConnectionManager.getConnection was called after the connection manager was closed!'));
}; };
return this.onProcessExit(); return this._onProcessExit();
} }
/**
* Initialize connection pool. By default pool autostart is set to false, so no connection will be
* be created unless `pool.acquire` is called.
*/
initPools() { initPools() {
const config = this.config; const config = this.config;
if (!config.replication) { if (!config.replication) {
this.pool = Pooling.createPool({ this.pool = Pooling.createPool({
create: () => new Promise(resolve => { create: () => this._connect(config).catch(err => err),
this destroy: mayBeConnection => {
._connect(config) if (mayBeConnection instanceof Error) {
.tap(() => { return Promise.resolve();
this.poolError = null; }
})
.then(resolve) return this._disconnect(mayBeConnection)
.catch(e => { .tap(() => { debug('connection destroy'); });
// dont throw otherwise pool will release _dispense call
// which will call _connect even if error is fatal
// https://github.com/coopernurse/node-pool/issues/161
this.poolError = e;
});
}),
destroy: connection => {
return this._disconnect(connection).tap(() => {
debug('connection destroy');
});
}, },
validate: config.pool.validate validate: config.pool.validate
}, { }, {
Promise: config.pool.Promise, Promise: config.pool.Promise,
max: config.pool.max,
min: config.pool.min,
testOnBorrow: true, testOnBorrow: true,
autostart: false, autostart: false,
max: config.pool.max,
min: config.pool.min,
acquireTimeoutMillis: config.pool.acquire, acquireTimeoutMillis: config.pool.acquire,
idleTimeoutMillis: config.pool.idle, idleTimeoutMillis: config.pool.idle,
evictionRunIntervalMillis: config.pool.evict evictionRunIntervalMillis: config.pool.evict
}); });
this.pool.on('factoryCreateError', error => { debug(`pool created with max/min: ${config.pool.max}/${config.pool.min}, no replication`);
this.poolError = error;
});
debug(`pool created max/min: ${config.pool.max}/${config.pool.min} with no replication`);
return; return;
} }
...@@ -150,21 +157,26 @@ class ConnectionManager { ...@@ -150,21 +157,26 @@ class ConnectionManager {
acquire: (priority, queryType, useMaster) => { acquire: (priority, queryType, useMaster) => {
useMaster = _.isUndefined(useMaster) ? false : useMaster; useMaster = _.isUndefined(useMaster) ? false : useMaster;
if (queryType === 'SELECT' && !useMaster) { if (queryType === 'SELECT' && !useMaster) {
return this.pool.read.acquire(priority); return this.pool.read.acquire(priority)
.then(mayBeConnection => this._determineConnection(mayBeConnection));
} else { } else {
return this.pool.write.acquire(priority); return this.pool.write.acquire(priority)
.then(mayBeConnection => this._determineConnection(mayBeConnection));
} }
}, },
destroy: connection => { destroy: mayBeConnection => {
debug('connection destroy'); if (mayBeConnection instanceof Error) {
return this.pool[connection.queryType].destroy(connection); return Promise.resolve();
}
this.pool[mayBeConnection.queryType].destroy(mayBeConnection)
.tap(() => { debug('connection destroy'); });
}, },
clear: () => { clear: () => {
debug('all connection clear');
return Promise.join( return Promise.join(
this.pool.read.clear(), this.pool.read.clear(),
this.pool.write.clear() this.pool.write.clear()
); ).tap(() => { debug('all connection clear'); });
}, },
drain: () => { drain: () => {
return Promise.join( return Promise.join(
...@@ -175,71 +187,62 @@ class ConnectionManager { ...@@ -175,71 +187,62 @@ class ConnectionManager {
read: Pooling.createPool({ read: Pooling.createPool({
create: () => { create: () => {
const nextRead = reads++ % config.replication.read.length; // round robin config const nextRead = reads++ % config.replication.read.length; // round robin config
return new Promise(resolve => { return this
this
._connect(config.replication.read[nextRead]) ._connect(config.replication.read[nextRead])
.tap(connection => { .tap(connection => {
connection.queryType = 'read'; connection.queryType = 'read';
this.poolError = null;
resolve(connection);
}) })
.catch(e => { .catch(err => err);
this.poolError = e;
});
});
},
destroy: connection => {
return this._disconnect(connection);
}, },
destroy: connection => this._disconnect(connection),
validate: config.pool.validate validate: config.pool.validate
}, { }, {
Promise: config.pool.Promise, Promise: config.pool.Promise,
max: config.pool.max,
min: config.pool.min,
testOnBorrow: true, testOnBorrow: true,
autostart: false, autostart: false,
max: config.pool.max,
min: config.pool.min,
acquireTimeoutMillis: config.pool.acquire, acquireTimeoutMillis: config.pool.acquire,
idleTimeoutMillis: config.pool.idle, idleTimeoutMillis: config.pool.idle,
evictionRunIntervalMillis: config.pool.evict evictionRunIntervalMillis: config.pool.evict
}), }),
write: Pooling.createPool({ write: Pooling.createPool({
create: () => new Promise(resolve => { create: () => {
this return this
._connect(config.replication.write) ._connect(config.replication.write)
.then(connection => { .tap(connection => {
connection.queryType = 'write'; connection.queryType = 'write';
this.poolError = null;
return resolve(connection);
}) })
.catch(e => { .catch(err => err);
this.poolError = e;
});
}),
destroy: connection => {
return this._disconnect(connection);
}, },
destroy: connection => this._disconnect(connection),
validate: config.pool.validate validate: config.pool.validate
}, { }, {
Promise: config.pool.Promise, Promise: config.pool.Promise,
max: config.pool.max,
min: config.pool.min,
testOnBorrow: true, testOnBorrow: true,
autostart: false, autostart: false,
max: config.pool.max,
min: config.pool.min,
acquireTimeoutMillis: config.pool.acquire, acquireTimeoutMillis: config.pool.acquire,
idleTimeoutMillis: config.pool.idle, idleTimeoutMillis: config.pool.idle,
evictionRunIntervalMillis: config.pool.evict evictionRunIntervalMillis: config.pool.evict
}) })
}; };
this.pool.read.on('factoryCreateError', error => { debug(`pool created with max/min: ${config.pool.max}/${config.pool.min}, with replication`);
this.poolError = error;
});
this.pool.write.on('factoryCreateError', error => {
this.poolError = error;
});
} }
/**
* Get connection from pool. It set database version if its not already set.
* Call pool.acquire to get a connection
*
* @param {Object} [options] Pool options
* @param {Integer} [options.priority] Set priority for this call. Read more at https://github.com/coopernurse/node-pool#priority-queueing
* @param {String} [options.type] Set which replica to use. Available options are `read` and `write`
* @param {Boolean} [options.useMaster=false] Force master or write replica to get connection from
*
* @return {Promise<Connection>}
*/
getConnection(options) { getConnection(options) {
options = options || {}; options = options || {};
...@@ -248,8 +251,10 @@ class ConnectionManager { ...@@ -248,8 +251,10 @@ class ConnectionManager {
if (this.versionPromise) { if (this.versionPromise) {
promise = this.versionPromise; promise = this.versionPromise;
} else { } else {
promise = this.versionPromise = this._connect(this.config.replication.write || this.config).then(connection => { promise = this.versionPromise = this._connect(this.config.replication.write || this.config)
.then(connection => {
const _options = {}; const _options = {};
_options.transaction = {connection}; // Cheat .query to use our private connection _options.transaction = {connection}; // Cheat .query to use our private connection
_options.logging = () => {}; _options.logging = () => {};
_options.logging.__testLoggingFn = true; _options.logging.__testLoggingFn = true;
...@@ -270,42 +275,79 @@ class ConnectionManager { ...@@ -270,42 +275,79 @@ class ConnectionManager {
} }
return promise.then(() => { return promise.then(() => {
return Promise.race([ return this.pool.acquire(options.priority, options.type, options.useMaster)
this.pool.acquire(options.priority, options.type, options.useMaster), .then(mayBeConnection => this._determineConnection(mayBeConnection))
new Promise((resolve, reject) => .tap(() => { debug('connection acquired'); });
timers.setTimeout(() => {
if (this.poolError) {
reject(this.poolError);
}
}, 0))
])
.tap(() => { debug('connection acquired'); })
.catch(e => {
e = this.poolError || e;
this.poolError = null;
throw e;
});
}); });
} }
/**
* Release a pooled connection so it can be utilized by other connection requests
*
* @param {Connection} connection
*
* @return {Promise}
*/
releaseConnection(connection) { releaseConnection(connection) {
return this.pool.release(connection).tap(() => { return this.pool.release(connection)
debug('connection released'); .tap(() => { debug('connection released'); })
}); .catch(/Resource not currently part of this pool/, () => {});
} }
/**
* Check if something acquired by pool is indeed a connection but not an Error instance
* Why we need to do this https://github.com/sequelize/sequelize/pull/8330
*
* @param {Object|Error} mayBeConnection Object which can be either connection or error
*
* @retun {Promise<Connection>}
*/
_determineConnection(mayBeConnection) {
if (mayBeConnection instanceof Error) {
return Promise.resolve(this.pool.destroy(mayBeConnection))
.catch(/Resource not currently part of this pool/, () => {})
.then(() => { throw mayBeConnection; });
}
return Promise.resolve(mayBeConnection);
}
/**
* Call dialect library to get connection
*
* @param {*} config Connection config
* @private
* @return {Promise<Connection>}
*/
_connect(config) { _connect(config) {
return this.sequelize.runHooks('beforeConnect', config) return this.sequelize.runHooks('beforeConnect', config)
.then(() => this.dialect.connectionManager.connect(config)) .then(() => this.dialect.connectionManager.connect(config))
.then(connection => this.sequelize.runHooks('afterConnect', connection, config).return(connection)); .then(connection => this.sequelize.runHooks('afterConnect', connection, config).return(connection));
} }
/**
* Call dialect library to disconnect a connection
*
* @param {Connection} connection
* @private
* @return {Promise}
*/
_disconnect(connection) { _disconnect(connection) {
return this.dialect.connectionManager.disconnect(connection); return this.dialect.connectionManager.disconnect(connection);
} }
/**
* Determine if a connection is still valid or not
*
* @param {Connection} connection
*
* @return {Boolean}
*/
_validate(connection) { _validate(connection) {
if (!this.dialect.connectionManager.validate) return true; if (!this.dialect.connectionManager.validate) {
return true;
}
return this.dialect.connectionManager.validate(connection); return this.dialect.connectionManager.validate(connection);
} }
} }
......
lib/dialects/mssql/connection-manager.js
...@@ -121,7 +121,8 @@ class ConnectionManager extends AbstractConnectionManager { ...@@ -121,7 +121,8 @@ class ConnectionManager extends AbstractConnectionManager {
switch (err.code) { switch (err.code) {
case 'ESOCKET': case 'ESOCKET':
case 'ECONNRESET': case 'ECONNRESET':
this.pool.destroy(connectionLock); this.pool.destroy(connectionLock)
.catch(/Resource not currently part of this pool/, () => {});
} }
}); });
} }
......
lib/dialects/mysql/connection-manager.js
...@@ -99,6 +99,8 @@ class ConnectionManager extends AbstractConnectionManager { ...@@ -99,6 +99,8 @@ class ConnectionManager extends AbstractConnectionManager {
debug(`connection error ${e.code}`); debug(`connection error ${e.code}`);
if (e.code === 'PROTOCOL_CONNECTION_LOST') { if (e.code === 'PROTOCOL_CONNECTION_LOST') {
this.pool.destroy(connection)
.catch(/Resource not currently part of this pool/, () => {});
return; return;
} }
} }
......
lib/dialects/postgres/connection-manager.js
...@@ -35,7 +35,6 @@ class ConnectionManager extends AbstractConnectionManager { ...@@ -35,7 +35,6 @@ class ConnectionManager extends AbstractConnectionManager {
// Expose this as a method so that the parsing may be updated when the user has added additional, custom types // Expose this as a method so that the parsing may be updated when the user has added additional, custom types
_refreshTypeParser(dataType) { _refreshTypeParser(dataType) {
if (dataType.types.postgres.oids) { if (dataType.types.postgres.oids) {
for (const oid of dataType.types.postgres.oids) { for (const oid of dataType.types.postgres.oids) {
this.lib.types.setTypeParser(oid, value => dataType.parse(value, oid, this.lib.types.getTypeParser)); this.lib.types.setTypeParser(oid, value => dataType.parse(value, oid, this.lib.types.getTypeParser));
...@@ -54,7 +53,6 @@ class ConnectionManager extends AbstractConnectionManager { ...@@ -54,7 +53,6 @@ class ConnectionManager extends AbstractConnectionManager {
} }
connect(config) { connect(config) {
config.user = config.username; config.user = config.username;
const connectionConfig = Utils._.pick(config, [ const connectionConfig = Utils._.pick(config, [
'user', 'password', 'host', 'database', 'port' 'user', 'password', 'host', 'database', 'port'
...@@ -174,6 +172,7 @@ class ConnectionManager extends AbstractConnectionManager { ...@@ -174,6 +172,7 @@ class ConnectionManager extends AbstractConnectionManager {
}); });
}); });
} }
disconnect(connection) { disconnect(connection) {
return new Promise(resolve => { return new Promise(resolve => {
connection.end(); connection.end();
......
lib/sequelize.js
...@@ -82,12 +82,12 @@ class Sequelize { ...@@ -82,12 +82,12 @@ class Sequelize {
* @param {Object} [options.pool] sequelize connection pool configuration * @param {Object} [options.pool] sequelize connection pool configuration
* @param {Integer} [options.pool.max=5] Maximum number of connection in pool * @param {Integer} [options.pool.max=5] Maximum number of connection in pool
* @param {Integer} [options.pool.min=0] Minimum number of connection in pool * @param {Integer} [options.pool.min=0] Minimum number of connection in pool
* @param {Integer} [options.pool.idle=10000] The maximum time, in milliseconds, that a connection can be idle before being released * @param {Integer} [options.pool.idle=10000] The maximum time, in milliseconds, that a connection can be idle before being released. Use with combination of evict for proper working, for more details read https://github.com/coopernurse/node-pool/issues/178#issuecomment-327110870
* @param {Integer} [options.pool.acquire=10000] The maximum time, in milliseconds, that pool will try to get connection before throwing error * @param {Integer} [options.pool.acquire=10000] The maximum time, in milliseconds, that pool will try to get connection before throwing error
* @param {Integer} [options.pool.evict=60000] The time interval, in milliseconds, for evicting stale connections. Set it to 0 to disable this feature. * @param {Integer} [options.pool.evict=10000] The time interval, in milliseconds, for evicting stale connections. Set it to 0 to disable this feature.
* @param {Boolean} [options.pool.handleDisconnects=true] Controls if pool should handle connection disconnect automatically without throwing errors * @param {Boolean} [options.pool.handleDisconnects=true] Controls if pool should handle connection disconnect automatically without throwing errors
* @param {Function} [options.pool.validate] A function that validates a connection. Called with client. The default function checks that client is an object, and that its state is not disconnected * @param {Function} [options.pool.validate] A function that validates a connection. Called with client. The default function checks that client is an object, and that its state is not disconnected
* @param {Boolean} [options.quoteIdentifiers=true] Set to `false` to make table names and attributes case-insensitive on Postgres and skip double quoting of them. WARNING: Setting this to false may expose vulnerabilities and is not reccomended! * @param {Boolean} [options.quoteIdentifiers=true] Set to `false` to make table names and attributes case-insensitive on Postgres and skip double quoting of them. WARNING: Setting this to false may expose vulnerabilities and is not recommended!
* @param {String} [options.transactionType='DEFERRED'] Set the default transaction type. See `Sequelize.Transaction.TYPES` for possible options. Sqlite only. * @param {String} [options.transactionType='DEFERRED'] Set the default transaction type. See `Sequelize.Transaction.TYPES` for possible options. Sqlite only.
* @param {String} [options.isolationLevel] Set the default transaction isolation level. See `Sequelize.Transaction.ISOLATION_LEVELS` for possible options. * @param {String} [options.isolationLevel] Set the default transaction isolation level. See `Sequelize.Transaction.ISOLATION_LEVELS` for possible options.
* @param {Object} [options.retry] Set of flags that control when a query is automatically retried. * @param {Object} [options.retry] Set of flags that control when a query is automatically retried.
...@@ -315,7 +315,7 @@ class Sequelize { ...@@ -315,7 +315,7 @@ class Sequelize {
* *
* sequelize.models.modelName // The model will now be available in models under the name given to define * sequelize.models.modelName // The model will now be available in models under the name given to define
*/ */
define(modelName, attributes, options) { // testhint options:none define(modelName, attributes, options) {
options = options || {}; options = options || {};
options.modelName = modelName; options.modelName = modelName;
...@@ -325,7 +325,6 @@ class Sequelize { ...@@ -325,7 +325,6 @@ class Sequelize {
model.init(attributes, options); model.init(attributes, options);
return model; return model;
} }
...@@ -377,7 +376,7 @@ class Sequelize { ...@@ -377,7 +376,7 @@ class Sequelize {
if (!this.importCache[path]) { if (!this.importCache[path]) {
let defineCall = arguments.length > 1 ? arguments[1] : require(path); let defineCall = arguments.length > 1 ? arguments[1] : require(path);
if (typeof defineCall === 'object') { if (typeof defineCall === 'object') {
// ES6 module compatability // ES6 module compatibility
defineCall = defineCall.default; defineCall = defineCall.default;
} }
this.importCache[path] = defineCall(this, DataTypes); this.importCache[path] = defineCall(this, DataTypes);
...@@ -656,7 +655,7 @@ class Sequelize { ...@@ -656,7 +655,7 @@ class Sequelize {
* @param {Boolean|function} [options.logging=console.log] A function that logs sql queries, or false for no logging * @param {Boolean|function} [options.logging=console.log] A function that logs sql queries, or false for no logging
* @param {String} [options.schema='public'] The schema that the tables should be created in. This can be overriden for each table in sequelize.define * @param {String} [options.schema='public'] The schema that the tables should be created in. This can be overriden for each table in sequelize.define
* @param {String} [options.searchPath=DEFAULT] An optional parameter to specify the schema search_path (Postgres only) * @param {String} [options.searchPath=DEFAULT] An optional parameter to specify the schema search_path (Postgres only)
* @param {Boolean} [options.hooks=true] If hooks is true then beforeSync, afterSync, beforBulkSync, afterBulkSync hooks will be called * @param {Boolean} [options.hooks=true] If hooks is true then beforeSync, afterSync, beforeBulkSync, afterBulkSync hooks will be called
* @param {Boolean} [options.alter=false] Alters tables to fit models. Not recommended for production use. Deletes data in columns that were removed or had their type changed in the model. * @param {Boolean} [options.alter=false] Alters tables to fit models. Not recommended for production use. Deletes data in columns that were removed or had their type changed in the model.
* @return {Promise} * @return {Promise}
*/ */
...@@ -950,7 +949,6 @@ class Sequelize { ...@@ -950,7 +949,6 @@ class Sequelize {
autoCallback = options; autoCallback = options;
options = undefined; options = undefined;
} }
// testhint argsConform.end
const transaction = new Transaction(this, options); const transaction = new Transaction(this, options);
...@@ -1061,9 +1059,11 @@ class Sequelize { ...@@ -1061,9 +1059,11 @@ class Sequelize {
* *
* Normally this is done on process exit, so you only need to call this method if you are creating multiple instances, and want * Normally this is done on process exit, so you only need to call this method if you are creating multiple instances, and want
* to garbage collect some of them. * to garbage collect some of them.
*
* @return {Promise}
*/ */
close() { close() {
this.connectionManager.close(); return this.connectionManager.close();
} }
normalizeDataType(Type) { normalizeDataType(Type) {
......
test/config/.docker.env
...@@ -5,3 +5,7 @@ SEQ_MYSQL_PW=sequelize_test ...@@ -5,3 +5,7 @@ SEQ_MYSQL_PW=sequelize_test
SEQ_PG_PORT=8998 SEQ_PG_PORT=8998
SEQ_PG_USER=sequelize_test SEQ_PG_USER=sequelize_test
SEQ_PG_PW=sequelize_test SEQ_PG_PW=sequelize_test
SEQ_MSSQL_PORT=8997
SEQ_MSSQL_DB=master
SEQ_MSSQL_USER=sa
SEQ_MSSQL_PW=yourStrong(!)Password
test/config/config.js
...@@ -26,7 +26,7 @@ module.exports = { ...@@ -26,7 +26,7 @@ module.exports = {
database: process.env.SEQ_MSSQL_DB || process.env.SEQ_DB || 'sequelize_test', database: process.env.SEQ_MSSQL_DB || process.env.SEQ_DB || 'sequelize_test',
username: process.env.SEQ_MSSQL_USER || process.env.SEQ_USER || 'sequelize', username: process.env.SEQ_MSSQL_USER || process.env.SEQ_USER || 'sequelize',
password: process.env.SEQ_MSSQL_PW || process.env.SEQ_PW || 'nEGkLma26gXVHFUAHJxcmsrK', password: process.env.SEQ_MSSQL_PW || process.env.SEQ_PW || 'nEGkLma26gXVHFUAHJxcmsrK',
host: process.env.SEQ_MSSQL_HOST || process.env.SEQ_HOST || 'mssql.sequelizejs.com', host: process.env.SEQ_MSSQL_HOST || process.env.SEQ_HOST || '127.0.0.1',
port: process.env.SEQ_MSSQL_PORT || process.env.SEQ_PORT || 1433, port: process.env.SEQ_MSSQL_PORT || process.env.SEQ_PORT || 1433,
dialectOptions: { dialectOptions: {
// big insert queries need a while // big insert queries need a while
......
test/integration/dialects/abstract/connection-manager.test.js
...@@ -18,7 +18,6 @@ const poolEntry = { ...@@ -18,7 +18,6 @@ const poolEntry = {
}; };
describe('Connection Manager', () => { describe('Connection Manager', () => {
let sandbox; let sandbox;
beforeEach(() => { beforeEach(() => {
...@@ -162,7 +161,6 @@ describe('Connection Manager', () => { ...@@ -162,7 +161,6 @@ describe('Connection Manager', () => {
expect(poolDrainSpy.calledOnce).to.be.true; expect(poolDrainSpy.calledOnce).to.be.true;
expect(poolClearSpy.calledOnce).to.be.true; expect(poolClearSpy.calledOnce).to.be.true;
}); });
}); });
}); });
test/integration/dialects/mssql/connection-manager.test.js
...@@ -56,6 +56,30 @@ if (dialect.match(/^mssql/)) { ...@@ -56,6 +56,30 @@ if (dialect.match(/^mssql/)) {
}); });
}); });
it('should not throw when non pooled connection is unexpectedly closed', () => {
const sequelize = Support.createSequelizeInstance({ pool: { min: 1, max: 1, idle: 5000 } });
const cm = sequelize.connectionManager;
let conn;
return sequelize
.sync()
.then(() => cm.getConnection())
.then(connection => {
conn = connection;
// remove from pool
return cm.pool.destroy(connection);
})
.then(() => {
// unexpected disconnect
const unwrapConn = conn.unwrap();
unwrapConn.emit('error', {
code: 'ESOCKET'
});
});
});
describe('Errors', () => { describe('Errors', () => {
it('ECONNREFUSED', () => { it('ECONNREFUSED', () => {
const sequelize = Support.createSequelizeInstance({ port: 34237 }); const sequelize = Support.createSequelizeInstance({ port: 34237 });
......
test/integration/dialects/mysql/connector-manager.test.js
...@@ -34,18 +34,20 @@ if (dialect === 'mysql') { ...@@ -34,18 +34,20 @@ if (dialect === 'mysql') {
it('accepts new queries after shutting down a connection', () => { it('accepts new queries after shutting down a connection', () => {
// Create a sequelize instance with fast disconnecting connection // Create a sequelize instance with fast disconnecting connection
const sequelize = Support.createSequelizeInstance({ pool: { idle: 50, max: 1 }}); const sequelize = Support.createSequelizeInstance({ pool: { idle: 50, max: 1, evict: 10 }});
const User = sequelize.define('User', { username: DataTypes.STRING }); const User = sequelize.define('User', { username: DataTypes.STRING });
return User return User
.sync({force: true}) .sync({force: true})
.then(() => User.create({username: 'user1'})) .then(() => User.create({ username: 'user1' }))
.then(() => sequelize.Promise.delay(100)) .then(() => sequelize.Promise.delay(100))
.then(() => { .then(() => {
// This query will be queued just after the `client.end` is executed and before its callback is called expect(sequelize.connectionManager.pool.size).to.equal(0);
//This query will be queued just after the `client.end` is executed and before its callback is called
return sequelize.query('SELECT COUNT(*) AS count FROM Users', { type: sequelize.QueryTypes.SELECT }); return sequelize.query('SELECT COUNT(*) AS count FROM Users', { type: sequelize.QueryTypes.SELECT });
}) })
.then(count => { .then(count => {
expect(sequelize.connectionManager.pool.size).to.equal(1);
expect(count[0].count).to.equal(1); expect(count[0].count).to.equal(1);
}); });
}); });
...@@ -68,7 +70,7 @@ if (dialect === 'mysql') { ...@@ -68,7 +70,7 @@ if (dialect === 'mysql') {
return cm.getConnection(); return cm.getConnection();
}) })
.then(connection => { .then(connection => {
// Old threadId should be different from current new one // Old threadId should be same as current connection
expect(conn.threadId).to.be.equal(connection.threadId); expect(conn.threadId).to.be.equal(connection.threadId);
expect(cm.validate(conn)).to.be.ok; expect(cm.validate(conn)).to.be.ok;
...@@ -77,7 +79,7 @@ if (dialect === 'mysql') { ...@@ -77,7 +79,7 @@ if (dialect === 'mysql') {
}); });
it('should work with handleDisconnects before release', () => { it('should work with handleDisconnects before release', () => {
const sequelize = Support.createSequelizeInstance({pool: {min: 1, max: 1, handleDisconnects: true, idle: 5000}}); const sequelize = Support.createSequelizeInstance({pool: { max: 1, min: 1, handleDisconnects: true, idle: 5000 }});
const cm = sequelize.connectionManager; const cm = sequelize.connectionManager;
let conn; let conn;
...@@ -89,8 +91,9 @@ if (dialect === 'mysql') { ...@@ -89,8 +91,9 @@ if (dialect === 'mysql') {
conn = connection; conn = connection;
// simulate a unexpected end from MySQL2 // simulate a unexpected end from MySQL2
conn.stream.emit('end'); conn.stream.emit('end');
return cm.releaseConnection(connection);
}) })
.then(() => cm.releaseConnection(conn))
.then(() => { .then(() => {
// Get next available connection // Get next available connection
return cm.getConnection(); return cm.getConnection();
...@@ -98,7 +101,9 @@ if (dialect === 'mysql') { ...@@ -98,7 +101,9 @@ if (dialect === 'mysql') {
.then(connection => { .then(connection => {
// Old threadId should be different from current new one // Old threadId should be different from current new one
expect(conn.threadId).to.not.be.equal(connection.threadId); expect(conn.threadId).to.not.be.equal(connection.threadId);
expect(cm.validate(conn)).to.not.be.ok; expect(sequelize.connectionManager.pool.size).to.equal(1);
expect(cm.validate(conn)).to.be.not.ok;
return cm.releaseConnection(connection); return cm.releaseConnection(connection);
}); });
}); });
......
test/integration/replication.test.js
...@@ -57,26 +57,22 @@ describe(Support.getTestDialectTeaser('Replication'), function() { ...@@ -57,26 +57,22 @@ describe(Support.getTestDialectTeaser('Replication'), function() {
it('should be able to make a write', () => { it('should be able to make a write', () => {
return this.User.create({ return this.User.create({
firstName: Math.random().toString() firstName: Math.random().toString()
}) }).then(expectWriteCalls);
.then(expectWriteCalls);
}); });
it('should be able to make a read', () => { it('should be able to make a read', () => {
return this.User.findAll() return this.User.findAll().then(expectReadCalls);
.then(expectReadCalls);
}); });
it('should run read-only transactions on the replica', () => { it('should run read-only transactions on the replica', () => {
return this.sequelize.transaction({readOnly: true}, transaction => { return this.sequelize.transaction({readOnly: true}, transaction => {
return this.User.findAll({transaction}); return this.User.findAll({transaction});
}) }).then(expectReadCalls);
.then(expectReadCalls);
}); });
it('should run non-read-only transactions on the primary', () => { it('should run non-read-only transactions on the primary', () => {
return this.sequelize.transaction(transaction => { return this.sequelize.transaction(transaction => {
return this.User.findAll({transaction}); return this.User.findAll({transaction});
}) }).then(expectWriteCalls);
.then(expectWriteCalls);
}); });
}); });
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!