From fd3e34b0cc98104239aea7c25555feec24d6a2b1 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Wed, 18 Nov 2015 00:11:54 -0500 Subject: [PATCH 1/9] Relay ctrlConnection errors on Zongji instance --- index.js | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/index.js b/index.js index b11842cf..a03cb546 100644 --- a/index.js +++ b/index.js @@ -4,6 +4,8 @@ var EventEmitter = require('events').EventEmitter; var generateBinlog = require('./lib/sequence/binlog'); function ZongJi(dsn, options) { + var self = this; + this.set(options); EventEmitter.call(this); @@ -12,6 +14,15 @@ function ZongJi(dsn, options) { var ctrlDsn = cloneObjectSimple(dsn); ctrlDsn.database = 'information_schema'; this.ctrlConnection = mysql.createConnection(ctrlDsn); + + this.ctrlConnection.on('error', function (error) { + self.emit('error', error); + }); + + this.ctrlConnection.on('unhandledError', function (error) { + self.emit('error', error); + }); + this.ctrlConnection.connect(); this.ctrlCallbacks = []; From 741e63235b279aa433f10d34a0a89dc77ad633e8 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Wed, 18 Nov 2015 00:13:20 -0500 Subject: [PATCH 2/9] Add unique, pk, constraintType and nullable to the table map --- index.js | 23 +++++++++++++++++++---- lib/binlog_event.js | 5 ++++- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/index.js b/index.js index a03cb546..f6e4ff55 100644 --- a/index.js +++ b/index.js @@ -158,10 +158,25 @@ ZongJi.prototype._executeCtrlCallbacks = function() { } }; -var tableInfoQueryTemplate = 'SELECT ' + - 'COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME, ' + - 'COLUMN_COMMENT, COLUMN_TYPE ' + - 'FROM columns ' + 'WHERE table_schema="%s" AND table_name="%s"'; +var tableInfoQueryTemplate = + ' SELECT c.COLUMN_NAME,' + + ' COLLATION_NAME,' + + ' CHARACTER_SET_NAME,' + + ' COLUMN_COMMENT,' + + ' COLUMN_TYPE,' + + ' IS_NULLABLE,' + + ' t.CONSTRAINT_TYPE' + + ' FROM information_schema.columns c' + + 'LEFT JOIN information_schema.KEY_COLUMN_USAGE k' + + ' ON k.TABLE_NAME = c.TABLE_NAME' + + ' AND k.TABLE_SCHEMA = c.TABLE_SCHEMA' + + ' AND c.COLUMN_NAME = k.COLUMN_NAME' + + 'LEFT JOIN information_schema.TABLE_CONSTRAINTS t' + + ' ON t.TABLE_NAME = c.TABLE_NAME' + + ' AND t.CONSTRAINT_SCHEMA = c.TABLE_SCHEMA' + + ' AND t.CONSTRAINT_NAME = k.CONSTRAINT_NAME' + + ' WHERE c.TABLE_SCHEMA = "%s"' + + ' AND c.TABLE_NAME = "%s";'.replace(/\s\s+/g, ' '); ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { var self = this; diff --git a/lib/binlog_event.js b/lib/binlog_event.js index 61ca95af..eb4f69f7 100644 --- a/lib/binlog_event.js +++ b/lib/binlog_event.js @@ -164,7 +164,10 @@ TableMap.prototype.updateColumnInfo = function() { name: columnSchemas[j].COLUMN_NAME, charset: columnSchemas[j].CHARACTER_SET_NAME, type: this.columnTypes[j], - // nullable: + constraintType: columnSchemas[j].CONSTRAINT_TYPE, + unique: columnSchemas[j].CONSTRAINT_TYPE === 'UNIQUE', + pk: columnSchemas[j].CONSTRAINT_TYPE === 'PRIMARY KEY', + nullable: columnSchemas[j].IS_NULLABLE, metadata: columnsMetadata[j] }); } From 35ebff673389acee0b6606db9a4d2dc8cd354e75 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Wed, 18 Nov 2015 00:18:26 -0500 Subject: [PATCH 3/9] Fix typo in SQL query --- index.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/index.js b/index.js index f6e4ff55..e29bef3e 100644 --- a/index.js +++ b/index.js @@ -170,7 +170,7 @@ var tableInfoQueryTemplate = 'LEFT JOIN information_schema.KEY_COLUMN_USAGE k' + ' ON k.TABLE_NAME = c.TABLE_NAME' + ' AND k.TABLE_SCHEMA = c.TABLE_SCHEMA' + - ' AND c.COLUMN_NAME = k.COLUMN_NAME' + + ' AND k.COLUMN_NAME = c.COLUMN_NAME' + 'LEFT JOIN information_schema.TABLE_CONSTRAINTS t' + ' ON t.TABLE_NAME = c.TABLE_NAME' + ' AND t.CONSTRAINT_SCHEMA = c.TABLE_SCHEMA' + From 395acf65ad2fa4b688b26b4414d944b4a793d950 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Wed, 18 Nov 2015 02:21:45 -0500 Subject: [PATCH 4/9] Include nullable and constraints in table map. --- index.js | 41 ++++++++++++++++++++++------------------- lib/binlog_event.js | 32 ++++++++++++++++++++++---------- 2 files changed, 44 insertions(+), 29 deletions(-) diff --git a/index.js b/index.js index e29bef3e..633ed68a 100644 --- a/index.js +++ b/index.js @@ -158,25 +158,28 @@ ZongJi.prototype._executeCtrlCallbacks = function() { } }; -var tableInfoQueryTemplate = - ' SELECT c.COLUMN_NAME,' + - ' COLLATION_NAME,' + - ' CHARACTER_SET_NAME,' + - ' COLUMN_COMMENT,' + - ' COLUMN_TYPE,' + - ' IS_NULLABLE,' + - ' t.CONSTRAINT_TYPE' + - ' FROM information_schema.columns c' + - 'LEFT JOIN information_schema.KEY_COLUMN_USAGE k' + - ' ON k.TABLE_NAME = c.TABLE_NAME' + - ' AND k.TABLE_SCHEMA = c.TABLE_SCHEMA' + - ' AND k.COLUMN_NAME = c.COLUMN_NAME' + - 'LEFT JOIN information_schema.TABLE_CONSTRAINTS t' + - ' ON t.TABLE_NAME = c.TABLE_NAME' + - ' AND t.CONSTRAINT_SCHEMA = c.TABLE_SCHEMA' + - ' AND t.CONSTRAINT_NAME = k.CONSTRAINT_NAME' + - ' WHERE c.TABLE_SCHEMA = "%s"' + - ' AND c.TABLE_NAME = "%s";'.replace(/\s\s+/g, ' '); +var tableInfoQueryTemplate = [ + ' SELECT c.COLUMN_NAME,', + ' COLLATION_NAME,', + ' CHARACTER_SET_NAME,', + ' COLUMN_COMMENT,', + ' COLUMN_TYPE,', + ' IS_NULLABLE,', + ' t.CONSTRAINT_TYPE,', + ' k.ORDINAL_POSITION,', + ' k.CONSTRAINT_NAME', + ' FROM information_schema.columns c', + 'LEFT JOIN information_schema.KEY_COLUMN_USAGE k', + ' ON k.TABLE_NAME = c.TABLE_NAME', + ' AND k.TABLE_SCHEMA = c.TABLE_SCHEMA', + ' AND k.COLUMN_NAME = c.COLUMN_NAME', + 'LEFT JOIN information_schema.TABLE_CONSTRAINTS t', + ' ON t.TABLE_NAME = c.TABLE_NAME', + ' AND t.CONSTRAINT_SCHEMA = c.TABLE_SCHEMA', + ' AND t.CONSTRAINT_NAME = k.CONSTRAINT_NAME', + ' WHERE c.TABLE_SCHEMA = "%s"', + ' AND c.TABLE_NAME = "%s"' +].join(' ').replace(/\s{2,}/g, ' '); ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { var self = this; diff --git a/lib/binlog_event.js b/lib/binlog_event.js index eb4f69f7..82b660d7 100644 --- a/lib/binlog_event.js +++ b/lib/binlog_event.js @@ -155,21 +155,33 @@ TableMap.prototype.updateColumnInfo = function() { delete columnsMetadata[i].type; } } + var tableMap = this.tableMap[this.tableId]; - var columnSchemas = tableMap.columnSchemas; var columns = []; + var column; + var columnSchema; + for (var j = 0; j < this.columnCount; j++) { - columns.push({ - name: columnSchemas[j].COLUMN_NAME, - charset: columnSchemas[j].CHARACTER_SET_NAME, + columnSchema = columnSchemas[j]; + + column = { + name: columnSchema.COLUMN_NAME, + charset: columnSchema.CHARACTER_SET_NAME, type: this.columnTypes[j], - constraintType: columnSchemas[j].CONSTRAINT_TYPE, - unique: columnSchemas[j].CONSTRAINT_TYPE === 'UNIQUE', - pk: columnSchemas[j].CONSTRAINT_TYPE === 'PRIMARY KEY', - nullable: columnSchemas[j].IS_NULLABLE, - metadata: columnsMetadata[j] - }); + nullable: columnSchema.IS_NULLABLE === 'YES', + metadata: columnsMetadata[j], + }; + + if (columnSchema.CONSTRAINT_TYPE) { + column.constraint = { + type: columnSchema.CONSTRAINT_TYPE, + position: columnSchema.ORDINAL_POSITION, + name: columnSchema.CONSTRAINT_NAME + }; + } + + columns.push(column); } tableMap.columns = columns; From ad732f6f1f579984adb8b13aef26ff230f373564 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Thu, 17 Dec 2015 15:52:21 -0500 Subject: [PATCH 5/9] Add appropriate error handling when insufficient permissions encountered while trying to read the information_schema --- index.js | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/index.js b/index.js index 633ed68a..ba9d54a4 100644 --- a/index.js +++ b/index.js @@ -190,6 +190,15 @@ ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { if (err) { // Errors should be emitted self.emit('error', err); + // This is a fatal error, no additional binlog events will be processed as next() will never be called + return; + } + + if (rows.length === 0) { + self.emit('error', + new Error('Insufficient permissions to access: ' + tableMapEvent.schemaName + '.' + tableMapEvent.tableName) + ); + // This is a fatal error, no additional binlog events will be processed as next() will never be called return; } @@ -275,8 +284,8 @@ ZongJi.prototype._skipSchema = function(database, table){ (include[database] instanceof Array && include[database].indexOf(table) !== -1)))) && (exclude === undefined || - (database !== undefined && - (!(database in exclude) || + (database !== undefined && + (!(database in exclude) || (exclude[database] !== true && (exclude[database] instanceof Array && exclude[database].indexOf(table) === -1)))))); From b0f2d591814eda42dd343e967aa2e55ed7228860 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Fri, 19 Feb 2016 11:57:14 -0500 Subject: [PATCH 6/9] Support columns with multiple constraints. Adds PK and UNIQUE boolean flags to each column. --- index.js | 40 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 36 insertions(+), 4 deletions(-) diff --git a/index.js b/index.js index ba9d54a4..063fa47b 100644 --- a/index.js +++ b/index.js @@ -159,14 +159,14 @@ ZongJi.prototype._executeCtrlCallbacks = function() { }; var tableInfoQueryTemplate = [ - ' SELECT c.COLUMN_NAME,', + ' SELECT c.ORDINAL_POSITION,' + + ' c.COLUMN_NAME,', ' COLLATION_NAME,', ' CHARACTER_SET_NAME,', ' COLUMN_COMMENT,', ' COLUMN_TYPE,', ' IS_NULLABLE,', ' t.CONSTRAINT_TYPE,', - ' k.ORDINAL_POSITION,', ' k.CONSTRAINT_NAME', ' FROM information_schema.columns c', 'LEFT JOIN information_schema.KEY_COLUMN_USAGE k', @@ -187,6 +187,12 @@ ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { tableMapEvent.schemaName, tableMapEvent.tableName); this.ctrlConnection.query(sql, function(err, rows) { + var columns = [], + column, + rowLen, + idx, + x; + if (err) { // Errors should be emitted self.emit('error', err); @@ -194,7 +200,9 @@ ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { return; } - if (rows.length === 0) { + rowLen = rows.length; + + if (rowLen === 0) { self.emit('error', new Error('Insufficient permissions to access: ' + tableMapEvent.schemaName + '.' + tableMapEvent.tableName) ); @@ -202,8 +210,32 @@ ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) { return; } + for (x = 0; x < rowLen; x++) { + column = rows[x]; + idx = column.ORDINAL_POSITION - 1; + + column.UNIQUE = column.CONSTRAINT_TYPE === 'UNIQUE'; + column.PK = column.CONSTRAINT_TYPE === 'PRIMARY KEY'; + + if (x === 0 || columns[idx] === undefined) { + columns.push(column); + } else { + columns[idx].CONSTRAINT_TYPES || (columns[idx].CONSTRAINT_TYPES = [columns[idx].CONSTRAINT_TYPE]); + columns[idx].CONSTRAINT_NAMES || (columns[idx].CONSTRAINT_NAMES = [columns[idx].CONSTRAINT_NAME]); + + columns[idx].CONSTRAINT_TYPES.push(column.CONSTRAINT_TYPE); + columns[idx].CONSTRAINT_NAMES.push(column.CONSTRAINT_NAME); + + if (column.UNIQUE) { + columns[idx].UNIQUE = true; + } else if (column.PK) { + columns[idx].PK = true; + } + } + } + self.tableMap[tableMapEvent.tableId] = { - columnSchemas: rows, + columnSchemas: columns, parentSchema: tableMapEvent.schemaName, tableName: tableMapEvent.tableName }; From 82512c1dd654d6f0c1b69fe39423b4e0c446ceac Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Fri, 19 Feb 2016 13:52:55 -0500 Subject: [PATCH 7/9] Zongji should support when the system timezone differs from the effective MySQL timezone --- index.js | 29 ++++++++++- lib/common.js | 137 +++++++++++++++++++++++++++++++++----------------- 2 files changed, 120 insertions(+), 46 deletions(-) diff --git a/index.js b/index.js index 063fa47b..6087bfe1 100644 --- a/index.js +++ b/index.js @@ -69,7 +69,14 @@ ZongJi.prototype._init = function() { binlogOptions.position = result.File_size; } } - } + }, + { + name: '_getUtcOffset', + callback: function(utcOffset) { + self.utcOffset = utcOffset * 1000; + require('./lib/common').setUtcOffset(self.utcOffset); + } + }, ]; var methodIndex = 0; @@ -99,6 +106,26 @@ ZongJi.prototype._init = function() { }; }; +ZongJi.prototype._getUtcOffset = function(next) { + var self = this; + var sql = 'SELECT TIMESTAMPDIFF(SECOND, NOW(), UTC_TIMESTAMP) AS offset;'; + var ctrlConnection = self.ctrlConnection; + + ctrlConnection.query(sql, function(err, rows) { + var nodeOffsetInSeconds = new Date().getTimezoneOffset() * 60, + mysqlOffsetInSeconds; + + if (err) { + self.emit('error', err); + return next(false); + } + + mysqlOffsetInSeconds = rows[0].offset; + + next(mysqlOffsetInSeconds - nodeOffsetInSeconds); + }); +}; + ZongJi.prototype._isChecksumEnabled = function(next) { var self = this; var sql = 'select @@GLOBAL.binlog_checksum as checksum'; diff --git a/lib/common.js b/lib/common.js index 57dc6924..59ed4bb4 100644 --- a/lib/common.js +++ b/lib/common.js @@ -30,9 +30,8 @@ var MysqlTypes = { VAR_STRING: 253, STRING: 254, GEOMETRY: 255 -}; - -var MysqlCodeTypes = { +}, +MysqlCodeTypes = { 0: 'DECIMAL', 1: 'TINY', 2: 'SHORT', @@ -63,9 +62,10 @@ var MysqlCodeTypes = { 253: 'VAR_STRING', 254: 'STRING', 255: 'GEOMETRY' -}; +}, +utcOffset = 0; -exports.parseUInt64 = function(parser) { +function parseUInt64(parser) { var low = parser.parseUnsignedNumber(4); var high = parser.parseUnsignedNumber(4); @@ -77,59 +77,59 @@ exports.parseUInt64 = function(parser) { // jshint bitwise: false return (high * Math.pow(2,32)) + low; -}; +} -exports.parseUInt48 = function(parser) { +function parseUInt48(parser) { var low = parser.parseUnsignedNumber(4); var high = parser.parseUnsignedNumber(2); // jshint bitwise: false return (high * Math.pow(2, 32)) + low; -}; +} -exports.parseUInt24 = function(parser) { +function parseUInt24(parser) { var low = parser.parseUnsignedNumber(2); var high = parser.parseUnsignedNumber(1); // jshint bitwise: false return (high << 16) + low; -}; +} -exports.parseBytesArray = function(parser, length) { +function parseBytesArray(parser, length) { var result = new Array(length); for (var i = 0; i < length; i++) { result[i] = parser.parseUnsignedNumber(1); } return result; -}; +} // Parse column definition list string for SET and ENUM data types // @param type String Definition of column 'set(...)' or 'enum(...)' // @param prefixLen Integer Number of characters before list starts // (e.g. 'set(': 4, 'enum(': 5) -var parseSetEnumTypeDef = function(type, prefixLen){ +function parseSetEnumTypeDef(type, prefixLen){ // listed distinct elements should not include commas return type.substr(prefixLen, type.length - prefixLen - 1) .split(',').map(function(opt){ return (opt[0] === '"' || opt[0] === "'") ? opt.substr(1, opt.length - 2) : opt; }); -}; +} function zeroPad(number, pad) { var N = Math.pow(10, pad); return number < N ? ('' + (N + number)).slice(1) : '' + number; } -var sliceBits = function(input, start, end) { +function sliceBits(input, start, end) { // ex: start: 10, end: 15 = "111110000000000" var match = (((1 << end) - 1) ^ ((1 << start) - 1)); return (input & match) >> start; -}; +} // See information about IEEE-754 Floating point numbers: // http://www.h-schmidt.net/FloatConverter/IEEE754.html // http://babbage.cs.qc.cuny.edu/IEEE-754.old/64bit.html // Pass only high for 32-bit float, pass high and low for 64-bit double -var parseIEEE754Float = function(high, low){ +function parseIEEE754Float(high, low){ var lastSignificantBit, sigFigs, expLeading; if(typeof low !== 'undefined'){ // 64-bit: 1 sign, 11 exponent, 52 significand @@ -164,15 +164,15 @@ var parseIEEE754Float = function(high, low){ } return sign * Math.pow(2, exponent) * significand; -}; +} -var getUInt32Value = function(input){ +function getUInt32Value(input){ // Last bit is not sign, it is part of value! if(input & (1 << 31)) return Math.pow(2, 31) + (input & ((1 << 31) -1)); else return input; -}; +} -var parseAnyInt = function(parser, column, columnSchema) { +function parseAnyInt(parser, column, columnSchema) { var result, int64, size; switch (column.type) { case MysqlTypes.TINY: @@ -185,7 +185,7 @@ var parseAnyInt = function(parser, column, columnSchema) { break; case MysqlTypes.INT24: size = 3; - result = exports.parseUInt24(parser); + result = - utcparseUInt24(parser); break; case MysqlTypes.LONG: size = 4; @@ -194,48 +194,47 @@ var parseAnyInt = function(parser, column, columnSchema) { case MysqlTypes.LONGLONG: size = 8; int64 = {}; - result = exports.parseUInt64.call(int64, parser); + result = parseUInt64.call(int64, parser); break; } - if(columnSchema.COLUMN_TYPE.indexOf('unsigned') === -1){ + if (columnSchema.COLUMN_TYPE.indexOf('unsigned') === -1) { var length = size * 8; // Flip bits on negative signed integer - if(!int64 && (result & (1 << (length - 1)))){ + if (!int64 && (result & (1 << (length - 1)))) { result = ((result ^ (Math.pow(2, length) - 1)) * -1) - 1; - }else if(int64 && (int64.high & (1 << 31))){ + } else if (int64 && (int64.high & (1 << 31))) { // Javascript integers only support 2^53 not 2^64, must trim bits! // 64-53 = 11, 32-11 = 21, so grab first 21 bits of high word only var mask = Math.pow(2, 32) - 1; var high = sliceBits(int64.high ^ mask, 0, 21); var low = int64.low ^ mask; - result = ((high * Math.pow(2, 32)) * - 1) - getUInt32Value(low) - 1; + result = ((high * Math.pow(2, 32)) * -1) - getUInt32Value(low) - 1; } } return result; -}; +} -var readInt24BE = function(buf, offset, noAssert) { +function readInt24BE(buf, offset, noAssert) { return (buf.readInt8(offset, noAssert) << 16) + buf.readUInt16BE(offset + 1, noAssert); -}; +} -var readIntBE = function(buf, offset, length, noAssert){ +function readIntBE(buf, offset, length, noAssert){ switch(length){ case 1: return buf.readInt8(offset, noAssert); case 2: return buf.readInt16BE(offset, noAssert); case 3: return readInt24BE(buf, offset, noAssert); case 4: return buf.readInt32BE(offset, noAssert); } -}; +} // Adapted from jeremycole's Ruby implementation: // https://github.com/jeremycole/mysql_binlog/blob/master/lib/mysql_binlog/binlog_field_parser.rb // Some more information about DECIMAL types: // http://dev.mysql.com/doc/refman/5.5/en/precision-math-decimal-characteristics.html -var parseNewDecimal = function(parser, column) { +function parseNewDecimal(parser, column) { // Constants of format var digitsPerInteger = 9; - var bytesPerInteger = 4; var compressedBytes = [0, 1, 1, 2, 2, 3, 3, 4, 4, 4]; var scale = column.metadata.decimals; @@ -289,11 +288,11 @@ var parseNewDecimal = function(parser, column) { } return parseFloat(str); -}; +} // Did not work in place. Function cribbed from lines 311-363 of // https://github.com/felixge/node-mysql/blob/cfd0ce3572d75c3c82103418d1d03cbe67eaf8a1/lib/protocol/Parser.js -var parseGeometryValue = function(buffer){ +function parseGeometryValue(buffer){ var offset = 4; if (buffer === null || !buffer.length) { @@ -347,15 +346,13 @@ var parseGeometryValue = function(buffer){ return result; } return parseGeometry(); -}; +} function convertToMysqlType (code) { return MysqlCodeTypes[code]; } -exports.convertToMysqlType = convertToMysqlType; - -var readTemporalFraction = function(parser, fractionPrecision) { +function readTemporalFraction(parser, fractionPrecision) { if(!fractionPrecision) return false; var fractionSize = Math.ceil(fractionPrecision / 2); var fraction = readIntBE(parser._buffer, parser._offset, fractionSize); @@ -379,7 +376,7 @@ var readTemporalFraction = function(parser, fractionPrecision) { }; }; -exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitter) { +function readMysqlValue(parser, column, columnSchema, tableMap, emitter) { var result; // jshint indent: false switch (column.type) { @@ -467,7 +464,7 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte result = parseGeometryValue(buffer); break; case MysqlTypes.DATE: - var raw = exports.parseUInt24(parser); + var raw = parseUInt24(parser); result = new Date( sliceBits(raw, 9, 24), // year sliceBits(raw, 5, 9) - 1, // month @@ -475,7 +472,7 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte ); break; case MysqlTypes.TIME: - var raw = exports.parseUInt24(parser); + var raw = parseUInt24(parser); var isNegative = (raw & (1 << 23)) !== 0; if(isNegative) raw = raw ^ ((1 << 24) - 1); // flip all bits @@ -516,7 +513,7 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte } break; case MysqlTypes.DATETIME: - var raw = exports.parseUInt64(parser); + var raw = parseUInt64(parser); var date = Math.floor(raw / 1000000); var time = raw % 1000000; result = new Date( @@ -527,6 +524,11 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte Math.floor((time % 10000) / 100), // minutes time % 100 // seconds ); + + if (utcOffset !== 0) { + result = new Date(result.getTime() - utcOffset); + } + break; case MysqlTypes.DATETIME2: // Overlapping high-low to get all data in 32-bit numbers @@ -545,10 +547,20 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte sliceBits(rawLow, 0, 6), // seconds fraction !== false ? fraction.milliseconds : 0 ); + + if (utcOffset !== 0) { + result = new Date(result.getTime() - utcOffset); + } + break; case MysqlTypes.TIMESTAMP: var raw = parser.parseUnsignedNumber(4); result = new Date(raw * 1000); + + if (utcOffset !== 0) { + result = new Date(result.getTime() - utcOffset()); + } + break; case MysqlTypes.TIMESTAMP2: var raw = readIntBE(parser._buffer, parser._offset, 4); @@ -556,6 +568,11 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte var fraction = readTemporalFraction(parser, column.metadata.decimals); var milliseconds = fraction !== false ? fraction.milliseconds : 0; result = new Date((raw * 1000) + milliseconds); + + if (utcOffset !== 0) { + result = new Date(result.getTime() - utcOffset); + } + break; case MysqlTypes.YEAR: var raw = parser.parseUnsignedNumber(1); @@ -578,4 +595,34 @@ exports.readMysqlValue = function(parser, column, columnSchema, tableMap, emitte '" in the database "' + tableMap.parentSchema + '"')); } return result; -}; +} + +function getUtcOffset() { + return utcOffset; +} + +function setUtcOffset(offset) { + utcOffset = offset; +} + +module.exports = { + parseUInt64: parseUInt64, + parseUInt48: parseUInt48, + parseUInt24: parseUInt24, + parseBytesArray: parseBytesArray, + parseSetEnumTypeDef: parseSetEnumTypeDef, + zeroPad: zeroPad, + sliceBits: sliceBits, + parseIEEE754Float: parseIEEE754Float, + getUInt32Value: getUInt32Value, + parseAnyInt: parseAnyInt, + readInt24BE: readInt24BE, + readIntBE: readIntBE, + parseNewDecimal: parseNewDecimal, + parseGeometryValue: parseGeometryValue, + convertToMysqlType: convertToMysqlType, + readTemporalFraction: readTemporalFraction, + readMysqlValue: readMysqlValue, + setUtcOffset: setUtcOffset, + getUtcOffset: getUtcOffset +}; \ No newline at end of file From 6beb055d4a21af882bbdeeac36b242e72d8024ad Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Thu, 3 Mar 2016 14:14:35 -0500 Subject: [PATCH 8/9] Make NPM complain less --- package.json | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/package.json b/package.json index 60d66262..0341812d 100644 --- a/package.json +++ b/package.json @@ -3,9 +3,6 @@ "version": "0.3.3", "description": "A mysql binlog listener running on Node.js", "main": "index.js", - "engines": { - "node": "0.10" - }, "directories": { "test": "test" }, @@ -33,6 +30,6 @@ "nodeunit": "~0.9.0" }, "dependencies": { - "mysql": "~2.9.0" + "mysql": "^2.10.2" } } From 58578a6323bcc33d7e2ede4f89662e7711206ae5 Mon Sep 17 00:00:00 2001 From: Jeffrey Mealo Date: Thu, 3 Mar 2016 14:33:18 -0500 Subject: [PATCH 9/9] Fix typo --- lib/common.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/common.js b/lib/common.js index 59ed4bb4..cf83d725 100644 --- a/lib/common.js +++ b/lib/common.js @@ -185,7 +185,7 @@ function parseAnyInt(parser, column, columnSchema) { break; case MysqlTypes.INT24: size = 3; - result = - utcparseUInt24(parser); + result = parseUInt24(parser); break; case MysqlTypes.LONG: size = 4; @@ -374,7 +374,7 @@ function readTemporalFraction(parser, fractionPrecision) { precision: fractionPrecision, milliseconds: milliseconds }; -}; +} function readMysqlValue(parser, column, columnSchema, tableMap, emitter) { var result;