diff --git a/index.js b/index.js index b11842cf..6087bfe1 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 = []; @@ -58,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; @@ -88,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'; @@ -147,10 +185,28 @@ 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.ORDINAL_POSITION,' + + ' c.COLUMN_NAME,', + ' COLLATION_NAME,', + ' CHARACTER_SET_NAME,', + ' COLUMN_COMMENT,', + ' COLUMN_TYPE,', + ' IS_NULLABLE,', + ' t.CONSTRAINT_TYPE,', + ' 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; @@ -158,14 +214,55 @@ 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); + // This is a fatal error, no additional binlog events will be processed as next() will never be called + return; + } + + rowLen = rows.length; + + if (rowLen === 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; } + 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 }; @@ -246,8 +343,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)))))); diff --git a/lib/binlog_event.js b/lib/binlog_event.js index 61ca95af..82b660d7 100644 --- a/lib/binlog_event.js +++ b/lib/binlog_event.js @@ -155,18 +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], - // 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; diff --git a/lib/common.js b/lib/common.js index 57dc6924..cf83d725 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 = parseUInt24(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); @@ -377,9 +374,9 @@ var readTemporalFraction = function(parser, fractionPrecision) { precision: fractionPrecision, milliseconds: milliseconds }; -}; +} -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 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" } }