Squashed 'zongji/' changes from a6a4d82..8f012db

8f012db TIMESTAMP2 data type added and tested
f39e91b DATETIME2 data type added and tested
93f498a TIME2 data type added and tested
57ca7ab Recognize checksum in row event
14dee38 Make querySequence callback asynchronous

git-subtree-dir: zongji
git-subtree-split: 8f012dbe1cc9aa04c900d62e578159bcc799b1eb

Author: numtel

.gitignore

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

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

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) {

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;

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;
+    }
   }
 }
 

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);
 
     });
   },

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) {

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);
         }
       });
     }