Merge commit '4f4a8f8f2d667471208557fa934d752b7371ea45'

Author: numtel

zongji/.gitignore

@@ -24,3 +24,5 @@ npm-install-stamp
 *.swo
 *.swp
 ben.js
+.vagrant
+Vagrantfile

zongji/README.md

@@ -5,6 +5,8 @@ ZongJi (踪迹) is pronounced as `zōng jì` in Chinese.
 
 This package is a "pure JS" implementation based on [`node-mysql`](https://github.com/felixge/node-mysql). Since v0.2.0, The native part (which was written in C++) has been dropped.
 
+This package has been tested with MySQL server 5.5.40 and 5.6.19. All MySQL server versions >= 5.1.15 are supported.
+
 ## Quick Start
 
 ```javascript
@@ -106,6 +108,7 @@ Name   | Description
 * :star2: [All types allowed by `node-mysql`](https://github.com/felixge/node-mysql#type-casting) are supported by this package.
 * :speak_no_evil: While 64-bit integers in MySQL (`BIGINT` type) allow values in the range of 2<sup>64</sup> (± ½ × 2<sup>64</sup> for signed values), Javascript's internal storage of numbers limits values to 2<sup>53</sup>, making the allowed range of `BIGINT` fields only `-9007199254740992` to `9007199254740992`. Unsigned 64-bit integers must also not exceed `9007199254740992`.
 * :point_right: `TRUNCATE` statement does not cause corresponding `DeleteRows` event. Use unqualified `DELETE FROM` for same effect.
+* When using fractional seconds with `DATETIME` and `TIMESTAMP` data types in MySQL > 5.6.4, only millisecond precision is available due to the limit of Javascript's `Date` object.
 
 ## Run Tests
 
@@ -124,6 +127,7 @@ I learnt many things from following resources while making ZongJi.
 * http://dev.mysql.com/doc/internals/en/replication-protocol.html
 * http://www.cs.wichita.edu/~chang/lecture/cs742/program/how-mysql-c-api.html
 * https://github.com/jeremycole/mysql_binlog (Ruby implemenation of MySQL binlog parser)
+* http://dev.mysql.com/doc/internals/en/date-and-time-data-type-representation.html
 
 ## License
 MIT

zongji/index.js

@@ -38,36 +38,54 @@ util.inherits(ZongJi, EventEmitter);
 
 ZongJi.prototype._init = function() {
   var self = this;
+  var binlogOptions = {
+    tableMap: self.tableMap,
+  };
 
-  this._isChecksumEnabled(function(checksumEnabled) {
-    self.useChecksum = checksumEnabled;
-    var options = {
-      tableMap: self.tableMap,
-      useChecksum: checksumEnabled,
-    };
-
-    if(self.options.serverId !== undefined){
-      options.serverId = self.options.serverId;
+  var asyncMethods = [
+    {
+      name: '_isChecksumEnabled',
+      callback: function(checksumEnabled) {
+        self.useChecksum = checksumEnabled;
+        binlogOptions.useChecksum = checksumEnabled
+      }
+    },
+    {
+      name: '_findBinlogEnd',
+      callback: function(result){
+        if(result && self.options.startAtEnd){
+          binlogOptions.filename = result.Log_name;
+          binlogOptions.position = result.File_size;
+        }
+      }
     }
+  ];
+
+  var methodIndex = 0;
+  var nextMethod = function(){
+    var method = asyncMethods[methodIndex];
+    self[method.name](function(/* args */){
+      method.callback.apply(this, arguments);
+      methodIndex++;
+      if(methodIndex < asyncMethods.length){
+        nextMethod();
+      }else{
+        ready();
+      }
+    });
+  };
+  nextMethod();
 
-    var ready = function(){
-      self.binlog = generateBinlog.call(self, options);
-      self.ready = true;
-      self._executeCtrlCallbacks();
+  var ready = function(){
+    // Run asynchronously from _init(), as serverId option set in start()
+    if(self.options.serverId !== undefined){
+      binlogOptions.serverId = self.options.serverId;
     }
 
-    if(self.options.startAtEnd){
-      self._findBinlogEnd(function(result){
-        if(result){
-          options.filename = result.Log_name;
-          options.position = result.File_size;
-        }
-        ready();
-      });
-    }else{
-      ready();
-    }
-  });
+    self.binlog = generateBinlog.call(self, binlogOptions);
+    self.ready = true;
+    self._executeCtrlCallbacks();
+  };
 };
 
 ZongJi.prototype._isChecksumEnabled = function(next) {

zongji/lib/common.js

@@ -16,6 +16,10 @@ var MysqlTypes = {
   'NEWDATE': 14,
   'VARCHAR': 15,
   'BIT': 16,
+  // Fractional temporal types in MySQL >=5.6.4
+  'TIMESTAMP2': 17,
+  'DATETIME2': 18,
+  'TIME2': 19,
   'NEWDECIMAL': 246,
   'ENUM': 247,
   'SET': 248,
@@ -81,6 +85,14 @@ var parseSetEnumTypeDef = function(type){
     });
 };
 
+// @param {string} type - ex. 'time(4)'
+// @return {int} - ex. 4
+var parseTypeDefIntArg = function(type){
+  var start = type.indexOf('(') + 1;
+  if(start === 0) return undefined;
+  return parseInt(type.substr(start, type.length - start - 1), 10);
+};
+
 var zeroPad = function(num, size) {
   // Max 32 digits
   var s = "00000000000000000000000000000000" + num;
@@ -328,6 +340,36 @@ var convertToMysqlType = exports.convertToMysqlType = function(code) {
   return result;
 };
 
+var readTemporalFraction = function(parser, columnSchema) {
+  var fractionPrecision = parseTypeDefIntArg(columnSchema.COLUMN_TYPE);
+  if(fractionPrecision === undefined || fractionPrecision === 0) return false;
+
+  var fractionSize = Math.ceil(fractionPrecision / 2);
+  var fraction = readIntBE(parser._buffer, parser._offset, fractionSize);
+  parser._offset += fractionSize;
+  if(fractionPrecision % 2 !== 0) fraction /= 10; // Not using full space
+  if(fraction < 0) fraction *= -1; // Negative time, fraction not negative
+
+  var fractionStr = zeroPad(fraction, fractionPrecision);
+  var milliseconds;
+  if(fractionStr.length > 3){
+    milliseconds = fractionStr.substr(0, 3);
+  }else if(fractionStr.length === 2){
+    milliseconds = fractionStr + '0';
+  }else if(fractionStr.length === 1){
+    milliseconds = fractionStr + '00';
+  }else{
+    milliseconds = fractionStr;
+  }
+
+  return {
+    value: fraction,
+    precision: fractionPrecision,
+    string: fractionStr,
+    milliseconds: parseInt(milliseconds, 10)
+  };
+};
+
 exports.readMysqlValue = function(parser, column, columnSchema) {
   var result;
   // jshint indent: false
@@ -431,11 +473,36 @@ exports.readMysqlValue = function(parser, column, columnSchema) {
       var minute = Math.floor((raw % 10000) / 100);
       var second = raw % 100;
       if(isNegative) second += 1;
+  
+      result = (isNegative ? '-' : '') +
+               zeroPad(hour, hour > 99 ? 3 : 2) + ':' +
+               zeroPad(minute, 2) + ':' +
+               zeroPad(second, 2);
+      break;      
+    case MysqlTypes.TIME2:
+      var raw = readIntBE(parser._buffer, parser._offset, 3);
+      parser._offset += 3;
+      var fraction = readTemporalFraction(parser, columnSchema);
+
+      var isNegative = (raw & (1 << 23)) === 0;
+      if(isNegative) raw = raw ^ ((1 << 24) - 1); // flip all bits
+
+      var hour = sliceBits(raw, 12, 22);
+      var minute = sliceBits(raw, 6, 12);
+      var second = sliceBits(raw, 0, 6);
+
+      if(isNegative && (fraction === false || fraction.value === 0)){
+        second++;
+      }
 
       result = (isNegative ? '-' : '') +
                zeroPad(hour, hour > 99 ? 3 : 2) + ':' +
                zeroPad(minute, 2) + ':' +
                zeroPad(second, 2);
+
+      if(fraction !== false){
+        result += '.' + fraction.string;
+      }
       break;      
     case MysqlTypes.DATETIME:
       var raw = exports.parseUInt64(parser);
@@ -450,10 +517,39 @@ exports.readMysqlValue = function(parser, column, columnSchema) {
         time % 100                            // seconds
       );
       break;
+    case MysqlTypes.DATETIME2:
+      // Overlapping high-low to get all data in 32-bit numbers
+      var rawHigh = readIntBE(parser._buffer, parser._offset, 4);
+      var rawLow = readIntBE(parser._buffer, parser._offset + 1, 4);
+      parser._offset += 5;
+      var fraction = readTemporalFraction(parser, columnSchema);
+
+      var yearMonth = sliceBits(rawHigh, 14, 31);
+      result = new Date(
+        Math.floor(yearMonth / 13), // year
+        (yearMonth % 13) - 1,       // month
+        sliceBits(rawLow, 17, 22),  // day
+        sliceBits(rawLow, 12, 17),  // hour
+        sliceBits(rawLow, 6, 12),   // minutes
+        sliceBits(rawLow, 0, 6)     // seconds
+      );
+
+      if(fraction !== false){
+        // Javascript Date only supports milliseconds
+        result.setMilliseconds(fraction.milliseconds);
+      }
+      break;
     case MysqlTypes.TIMESTAMP:
       var raw = parser.parseUnsignedNumber(4);
       result = new Date(raw * 1000);
       break;
+    case MysqlTypes.TIMESTAMP2:
+      var raw = readIntBE(parser._buffer, parser._offset, 4);
+      parser._offset += 4;
+      var fraction = readTemporalFraction(parser, columnSchema);
+      var milliseconds = fraction !== false ? fraction.milliseconds : 0;
+      result = new Date((raw * 1000) + milliseconds);
+      break;
     case MysqlTypes.YEAR:
       var raw = parser.parseUnsignedNumber(1);
       result = raw + 1900;

zongji/lib/rows_event.js

@@ -8,17 +8,21 @@ var Version2Events = [
   0x20, // DELETE_ROWS_EVENT_V2
 ];
 
+var CHECKSUM_SIZE = 4;
+
 /**
  * Generic RowsEvent class
  * Attributes:
  *   position: Position inside next binlog
  *   binlogName: Name of next binlog file
+ *   zongji: ZongJi instance
  **/
 
 function RowsEvent(parser, options, zongji) {
   BinlogEvent.apply(this, arguments);
   this._readTableId(parser);
   this.flags = parser.parseUnsignedNumber(2);
+  this.useChecksum = zongji.useChecksum;
 
   // Version 2 Events
   if (Version2Events.indexOf(options.eventType) !== -1) {
@@ -49,10 +53,21 @@ function RowsEvent(parser, options, zongji) {
       parser._offset += columnsPresentBitmapSize;
     }
 
+    if(this.useChecksum){
+      // Ignore the checksum at the end of this packet
+      parser._packetEnd -= CHECKSUM_SIZE;
+    }
+
     this.rows = [];
     while (!parser.reachedPacketEnd()) {
       this.rows.push(this._fetchOneRow(parser));
     }
+
+    if(this.useChecksum){
+      // Skip past the checksum at the end of the packet
+      parser._packetEnd += CHECKSUM_SIZE;
+      parser._offset += CHECKSUM_SIZE;
+    }
   }
 }
 

zongji/test/events.js

@@ -8,7 +8,7 @@ var conn = process.testZongJi || {};
 
 var checkTableMatches = function(tableName){
   return function(test, event){
-    var tableDetails = event.tableMap[event.tableId]; 
+    var tableDetails = event.tableMap[event.tableId];
     test.strictEqual(tableDetails.parentSchema, settings.database);
     test.strictEqual(tableDetails.tableName, tableName);
   };
@@ -65,15 +65,18 @@ module.exports = {
         includeEvents: ['tablemap', 'writerows']
       });
 
-      querySequence(conn.db, [
-        'INSERT INTO ' + conn.escId(testTable) + ' (col) VALUES (10)',
-      ], function(results){
-        // Should only have 2 events since ZongJi start
-        test.equal(events.length, 2);
-        test.equal(events[1].rows[0].col, 10);
-        zongji.stop();
-        test.done();
-      });
+      // Give enough time to initialize
+      setTimeout(function(){
+        querySequence(conn.db, [
+          'INSERT INTO ' + conn.escId(testTable) + ' (col) VALUES (10)',
+        ], function(results){
+          // Should only have 2 events since ZongJi start
+          test.equal(events.length, 2);
+          test.equal(events[1].rows[0].col, 10);
+          zongji.stop();
+          test.done();
+        });
+      }, 200);
 
     });
   },

zongji/test/helpers/connector.js

@@ -18,7 +18,16 @@ module.exports = function(settings, callback){
     'CREATE DATABASE ' + escId(settings.database),
     'USE ' + escId(settings.database),
     'RESET MASTER',
-  ], function(){
+    'SELECT VERSION() AS version'
+  ], function(results){
+    
+    self.mysqlVersion = results[results.length - 1][0].version
+      .split('-')[0]
+      .split('.')
+      .map(function(part){
+        return parseInt(part, 10);
+      });
+
     var zongji = self.zongji = new ZongJi(settings.connection);
 
     zongji.on('binlog', function(event) {

zongji/test/helpers/querySequence.js

@@ -19,7 +19,9 @@ module.exports = function(connection, debug, queries, callback){
         if(index < sequence.length - 1){
           sequence[index + 1]();
         }else{
-          callback(results);
+          setTimeout(function(){
+            callback(results);
+          }, 200);
         }
       });
     }