Skip to content

public / sequelize

Go to a project

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

Commit 722ed505 by Andy Edwards Committed by GitHub
Options

refactor(dialects/postgres): asyncify methods (#12129)

1 parent ceb0de26
Expand all
Inline Side-by-side
Showing with 163 additions and 170 deletions
lib/dialects/postgres/connection-manager.js
......@@ -84,7 +84,7 @@ class ConnectionManager extends AbstractConnectionManager {
return this.lib.types.getTypeParser(oid, ...args);
}
connect(config) {
async connect(config) {
config.user = config.username;
const connectionConfig = _.pick(config, [
'user', 'password', 'host', 'database', 'port'
......@@ -121,7 +121,7 @@ class ConnectionManager extends AbstractConnectionManager {
]));
}
return new Promise((resolve, reject) => {
const connection = await new Promise((resolve, reject) => {
let responded = false;
const connection = new this.lib.Client(connectionConfig);
......@@ -194,76 +194,71 @@ class ConnectionManager extends AbstractConnectionManager {
resolve(connection);
}
});
}).then(connection => {
let query = '';
if (this.sequelize.options.standardConformingStrings !== false && connection['standard_conforming_strings'] !== 'on') {
// Disable escape characters in strings
// see https://github.com/sequelize/sequelize/issues/3545 (security issue)
// see https://www.postgresql.org/docs/current/static/runtime-config-compatible.html#GUC-STANDARD-CONFORMING-STRINGS
query += 'SET standard_conforming_strings=on;';
}
});
if (this.sequelize.options.clientMinMessages !== false) {
query += `SET client_min_messages TO ${this.sequelize.options.clientMinMessages};`;
}
let query = '';
if (!this.sequelize.config.keepDefaultTimezone) {
const isZone = !!moment.tz.zone(this.sequelize.options.timezone);
if (isZone) {
query += `SET TIME ZONE '${this.sequelize.options.timezone}';`;
} else {
query += `SET TIME ZONE INTERVAL '${this.sequelize.options.timezone}' HOUR TO MINUTE;`;
}
}
if (this.sequelize.options.standardConformingStrings !== false && connection['standard_conforming_strings'] !== 'on') {
// Disable escape characters in strings
// see https://github.com/sequelize/sequelize/issues/3545 (security issue)
// see https://www.postgresql.org/docs/current/static/runtime-config-compatible.html#GUC-STANDARD-CONFORMING-STRINGS
query += 'SET standard_conforming_strings=on;';
}
if (query) {
return Promise.resolve(connection.query(query)).then(() => connection);
}
return connection;
}).then(connection => {
if (Object.keys(this.nameOidMap).length === 0 &&
this.enumOids.oids.length === 0 &&
this.enumOids.arrayOids.length === 0) {
return Promise.resolve(this._refreshDynamicOIDs(connection)).then(() => connection);
if (this.sequelize.options.clientMinMessages !== false) {
query += `SET client_min_messages TO ${this.sequelize.options.clientMinMessages};`;
}
if (!this.sequelize.config.keepDefaultTimezone) {
const isZone = !!moment.tz.zone(this.sequelize.options.timezone);
if (isZone) {
query += `SET TIME ZONE '${this.sequelize.options.timezone}';`;
} else {
query += `SET TIME ZONE INTERVAL '${this.sequelize.options.timezone}' HOUR TO MINUTE;`;
}
return connection;
}).then(connection => {
// Don't let a Postgres restart (or error) to take down the whole app
connection.on('error', error => {
connection._invalid = true;
debug(`connection error ${error.code || error.message}`);
this.pool.destroy(connection);
});
return connection;
}
if (query) {
await connection.query(query);
}
if (Object.keys(this.nameOidMap).length === 0 &&
this.enumOids.oids.length === 0 &&
this.enumOids.arrayOids.length === 0) {
await this._refreshDynamicOIDs(connection);
}
// Don't let a Postgres restart (or error) to take down the whole app
connection.on('error', error => {
connection._invalid = true;
debug(`connection error ${error.code || error.message}`);
this.pool.destroy(connection);
});
return connection;
}
disconnect(connection) {
async disconnect(connection) {
if (connection._ending) {
debug('connection tried to disconnect but was already at ENDING state');
return Promise.resolve();
return;
}
return promisify(callback => connection.end(callback))();
return await promisify(callback => connection.end(callback))();
}
validate(connection) {
return !connection._invalid && !connection._ending;
}
_refreshDynamicOIDs(connection) {
async _refreshDynamicOIDs(connection) {
const databaseVersion = this.sequelize.options.databaseVersion;
const supportedVersion = '8.3.0';
// Check for supported version
if ( (databaseVersion && semver.gte(databaseVersion, supportedVersion)) === false) {
return Promise.resolve();
return;
}
// Refresh dynamic OIDs for some types
// These include Geometry / Geography / HStore / Enum / Citext / Range
return (connection || this.sequelize).query(
const results = await (connection || this.sequelize).query(
'WITH ranges AS (' +
' SELECT pg_range.rngtypid, pg_type.typname AS rngtypname,' +
' pg_type.typarray AS rngtyparray, pg_range.rngsubtype' +
......@@ -273,46 +268,46 @@ class ConnectionManager extends AbstractConnectionManager {
' ranges.rngtypname, ranges.rngtypid, ranges.rngtyparray' +
' FROM pg_type LEFT OUTER JOIN ranges ON pg_type.oid = ranges.rngsubtype' +
' WHERE (pg_type.typtype IN(\'b\', \'e\'));'
).then(results => {
let result = Array.isArray(results) ? results.pop() : results;
// When searchPath is prepended then two statements are executed and the result is
// an array of those two statements. First one is the SET search_path and second is
// the SELECT query result.
if (Array.isArray(result)) {
if (result[0].command === 'SET') {
result = result.pop();
}
);
let result = Array.isArray(results) ? results.pop() : results;
// When searchPath is prepended then two statements are executed and the result is
// an array of those two statements. First one is the SET search_path and second is
// the SELECT query result.
if (Array.isArray(result)) {
if (result[0].command === 'SET') {
result = result.pop();
}
}
const newNameOidMap = {};
const newEnumOids = { oids: [], arrayOids: [] };
const newNameOidMap = {};
const newEnumOids = { oids: [], arrayOids: [] };
for (const row of result.rows) {
// Mapping enums, handled separatedly
if (row.typtype === 'e') {
newEnumOids.oids.push(row.oid);
if (row.typarray) newEnumOids.arrayOids.push(row.typarray);
continue;
}
for (const row of result.rows) {
// Mapping enums, handled separatedly
if (row.typtype === 'e') {
newEnumOids.oids.push(row.oid);
if (row.typarray) newEnumOids.arrayOids.push(row.typarray);
continue;
}
// Mapping base types and their arrays
newNameOidMap[row.typname] = { oid: row.oid };
if (row.typarray) newNameOidMap[row.typname].arrayOid = row.typarray;
// Mapping base types and their arrays
newNameOidMap[row.typname] = { oid: row.oid };
if (row.typarray) newNameOidMap[row.typname].arrayOid = row.typarray;
// Mapping ranges(of base types) and their arrays
if (row.rngtypid) {
newNameOidMap[row.typname].rangeOid = row.rngtypid;
if (row.rngtyparray) newNameOidMap[row.typname].arrayRangeOid = row.rngtyparray;
}
// Mapping ranges(of base types) and their arrays
if (row.rngtypid) {
newNameOidMap[row.typname].rangeOid = row.rngtypid;
if (row.rngtyparray) newNameOidMap[row.typname].arrayRangeOid = row.rngtyparray;
}
}
// Replace all OID mappings. Avoids temporary empty OID mappings.
this.nameOidMap = newNameOidMap;
this.enumOids = newEnumOids;
// Replace all OID mappings. Avoids temporary empty OID mappings.
this.nameOidMap = newNameOidMap;
this.enumOids = newEnumOids;
this.refreshTypeParser(dataTypes.postgres);
});
this.refreshTypeParser(dataTypes.postgres);
}
_clearDynamicOIDs() {
......
lib/dialects/postgres/query-interface.js
......@@ -26,7 +26,7 @@ const _ = require('lodash');
* @returns {Promise}
* @private
*/
function ensureEnums(qi, tableName, attributes, options, model) {
async function ensureEnums(qi, tableName, attributes, options, model) {
const keys = Object.keys(attributes);
const keyLen = keys.length;
......@@ -50,108 +50,106 @@ function ensureEnums(qi, tableName, attributes, options, model) {
}
}
return Promise.all(promises).then(results => {
promises = [];
let enumIdx = 0;
// This little function allows us to re-use the same code that prepends or appends new value to enum array
const addEnumValue = (field, value, relativeValue, position = 'before', spliceStart = promises.length) => {
const valueOptions = _.clone(options);
valueOptions.before = null;
valueOptions.after = null;
switch (position) {
case 'after':
valueOptions.after = relativeValue;
break;
case 'before':
default:
valueOptions.before = relativeValue;
break;
}
const results = await Promise.all(promises);
promises = [];
let enumIdx = 0;
// This little function allows us to re-use the same code that prepends or appends new value to enum array
const addEnumValue = (field, value, relativeValue, position = 'before', spliceStart = promises.length) => {
const valueOptions = _.clone(options);
valueOptions.before = null;
valueOptions.after = null;
switch (position) {
case 'after':
valueOptions.after = relativeValue;
break;
case 'before':
default:
valueOptions.before = relativeValue;
break;
}
promises.splice(spliceStart, 0, () => {
return qi.sequelize.query(qi.QueryGenerator.pgEnumAdd(
tableName, field, value, valueOptions
), valueOptions);
});
};
for (i = 0; i < keyLen; i++) {
const attribute = attributes[keys[i]];
const type = attribute.type;
const enumType = type.type || type;
const field = attribute.field || keys[i];
if (
type instanceof DataTypes.ENUM ||
type instanceof DataTypes.ARRAY && enumType instanceof DataTypes.ENUM //ARRAY sub type is ENUM
) {
// If the enum type doesn't exist then create it
if (!results[enumIdx]) {
promises.push(() => {
return qi.sequelize.query(qi.QueryGenerator.pgEnum(tableName, field, enumType, options), Object.assign({}, options, { raw: true }));
});
} else if (!!results[enumIdx] && !!model) {
const enumVals = qi.QueryGenerator.fromArray(results[enumIdx].enum_value);
const vals = enumType.values;
// Going through already existing values allows us to make queries that depend on those values
// We will prepend all new values between the old ones, but keep in mind - we can't change order of already existing values
// Then we append the rest of new values AFTER the latest already existing value
// E.g.: [1,2] -> [0,2,1] ==> [1,0,2]
// E.g.: [1,2,3] -> [2,1,3,4] ==> [1,2,3,4]
// E.g.: [1] -> [0,2,3] ==> [1,0,2,3]
let lastOldEnumValue;
let rightestPosition = -1;
for (let oldIndex = 0; oldIndex < enumVals.length; oldIndex++) {
const enumVal = enumVals[oldIndex];
const newIdx = vals.indexOf(enumVal);
lastOldEnumValue = enumVal;
if (newIdx === -1) {
continue;
}
promises.splice(spliceStart, 0, () => {
return qi.sequelize.query(qi.QueryGenerator.pgEnumAdd(
tableName, field, value, valueOptions
), valueOptions);
});
};
const newValuesBefore = vals.slice(0, newIdx);
const promisesLength = promises.length;
// we go in reverse order so we could stop when we meet old value
for (let reverseIdx = newValuesBefore.length - 1; reverseIdx >= 0; reverseIdx--) {
if (~enumVals.indexOf(newValuesBefore[reverseIdx])) {
break;
}
for (i = 0; i < keyLen; i++) {
const attribute = attributes[keys[i]];
const type = attribute.type;
const enumType = type.type || type;
const field = attribute.field || keys[i];
addEnumValue(field, newValuesBefore[reverseIdx], lastOldEnumValue, 'before', promisesLength);
}
if (
type instanceof DataTypes.ENUM ||
type instanceof DataTypes.ARRAY && enumType instanceof DataTypes.ENUM //ARRAY sub type is ENUM
) {
// If the enum type doesn't exist then create it
if (!results[enumIdx]) {
promises.push(() => {
return qi.sequelize.query(qi.QueryGenerator.pgEnum(tableName, field, enumType, options), Object.assign({}, options, { raw: true }));
});
} else if (!!results[enumIdx] && !!model) {
const enumVals = qi.QueryGenerator.fromArray(results[enumIdx].enum_value);
const vals = enumType.values;
// Going through already existing values allows us to make queries that depend on those values
// We will prepend all new values between the old ones, but keep in mind - we can't change order of already existing values
// Then we append the rest of new values AFTER the latest already existing value
// E.g.: [1,2] -> [0,2,1] ==> [1,0,2]
// E.g.: [1,2,3] -> [2,1,3,4] ==> [1,2,3,4]
// E.g.: [1] -> [0,2,3] ==> [1,0,2,3]
let lastOldEnumValue;
let rightestPosition = -1;
for (let oldIndex = 0; oldIndex < enumVals.length; oldIndex++) {
const enumVal = enumVals[oldIndex];
const newIdx = vals.indexOf(enumVal);
lastOldEnumValue = enumVal;
if (newIdx === -1) {
continue;
}
// we detect the most 'right' position of old value in new enum array so we can append new values to it
if (newIdx > rightestPosition) {
rightestPosition = newIdx;
const newValuesBefore = vals.slice(0, newIdx);
const promisesLength = promises.length;
// we go in reverse order so we could stop when we meet old value
for (let reverseIdx = newValuesBefore.length - 1; reverseIdx >= 0; reverseIdx--) {
if (~enumVals.indexOf(newValuesBefore[reverseIdx])) {
break;
}
addEnumValue(field, newValuesBefore[reverseIdx], lastOldEnumValue, 'before', promisesLength);
}
if (lastOldEnumValue && rightestPosition < vals.length - 1) {
const remainingEnumValues = vals.slice(rightestPosition + 1);
for (let reverseIdx = remainingEnumValues.length - 1; reverseIdx >= 0; reverseIdx--) {
addEnumValue(field, remainingEnumValues[reverseIdx], lastOldEnumValue, 'after');
}
// we detect the most 'right' position of old value in new enum array so we can append new values to it
if (newIdx > rightestPosition) {
rightestPosition = newIdx;
}
}
enumIdx++;
if (lastOldEnumValue && rightestPosition < vals.length - 1) {
const remainingEnumValues = vals.slice(rightestPosition + 1);
for (let reverseIdx = remainingEnumValues.length - 1; reverseIdx >= 0; reverseIdx--) {
addEnumValue(field, remainingEnumValues[reverseIdx], lastOldEnumValue, 'after');
}
}
enumIdx++;
}
}
}
return promises
.reduce((promise, asyncFunction) => promise.then(asyncFunction), Promise.resolve())
.then(result => {
// If ENUM processed, then refresh OIDs
if (promises.length) {
return Promise.resolve(qi.sequelize.dialect.connectionManager._refreshDynamicOIDs()).then(() => result);
}
return result;
});
});
const result = await promises
.reduce(async (promise, asyncFunction) => await asyncFunction(await promise), Promise.resolve());
// If ENUM processed, then refresh OIDs
if (promises.length) {
await qi.sequelize.dialect.connectionManager._refreshDynamicOIDs();
}
return result;
}
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!