Fix schema filtering

index.js

@@ -154,7 +154,8 @@ ZongJi.prototype.start = function(options) {
     self.connection._implyConnect();
     self.connection._protocol._enqueue(new self.binlog(function(error, event){
       if(error) return self.emit('error', error);
-      if(event === undefined) return; // Filtered out
+      // Do not emit events that have been filtered out
+      if(event === undefined || event._filtered === true) return;
 
       if (event.getTypeName() === 'TableMap') {
         var tableMap = self.tableMap[event.tableId];

lib/binlog_event.js

@@ -113,7 +113,7 @@ Query.prototype.dump = function() {
  * see http://dev.mysql.com/doc/internals/en/table-map-event.html
  **/
 
-function TableMap(parser, options) {
+function TableMap(parser, options, zongji) {
   BinlogEvent.apply(this, arguments);
   this.tableMap = options.tableMap;
 
@@ -128,14 +128,21 @@ function TableMap(parser, options) {
 
   var tableNameLength = parser.parseUnsignedNumber(1);
   this.tableName = parser.parseString(tableNameLength);
-  parser.parseUnsignedNumber(1);
 
-  this.columnCount = parser.parseLengthCodedNumber();
-  this.columnTypes = Common.parseBytesArray(parser, this.columnCount);
-  // column meta data length
-  parser.parseLengthCodedNumber();
-  this._readColumnMetadata(parser);
-  // ignore the rest
+  if(zongji._skipSchema(this.schemaName, this.tableName)){
+    // This event has been filtered out because of its database/table
+    parser._offset = parser._packetEnd;
+    this._filtered = true;
+  }else{
+    parser.parseUnsignedNumber(1);
+
+    this.columnCount = parser.parseLengthCodedNumber();
+    this.columnTypes = Common.parseBytesArray(parser, this.columnCount);
+    // column meta data length
+    parser.parseLengthCodedNumber();
+    this._readColumnMetadata(parser);
+    // ignore the rest
+  }
 }
 
 util.inherits(TableMap, BinlogEvent);

lib/rows_event.js

@@ -32,12 +32,14 @@ function RowsEvent(parser, options, zongji) {
 
   this.tableMap = options.tableMap;
 
-  var database = this.tableMap[this.tableId].parentSchema;
-  var table = this.tableMap[this.tableId].tableName;
-  if(zongji._skipSchema(database, table)){
-    // This event has been filtered out because of its database/table
+  var tableData = this.tableMap[this.tableId];
+  if(tableData === undefined){
+    // TableMap event was filtered
     parser._offset = parser._packetEnd;
+    this._filtered = true;
   }else{
+    var database = tableData.parentSchema;
+    var table = tableData.tableName;
     var columnsPresentBitmapSize = Math.floor((this.numberOfColumns + 7) / 8);
     // Columns present bitmap exceeds 4 bytes with >32 rows
     // And is not handled anyways so just skip over its space

test/filtering.js

@@ -1,9 +1,6 @@
 var settings = require('./settings/mysql');
 var connector =  require('./helpers/connector');
-
-var codeMap = require('../lib/code_map');
-var initBinlogHeader = require('../lib/packet/binlog_header');
-var rowEvents = require('../lib/rows_event');
+var querySequence = require('./helpers/querySequence');
 
 var conn = process.testZongJi || {};
 
@@ -13,9 +10,11 @@ module.exports = {
     else done();
   },
   tearDown: function(done){
+    conn && conn.eventLog.splice(0, conn.eventLog.length);
+    conn && conn.errorLog.splice(0, conn.errorLog.length);
     done();
   },
-  testFilter: function(test){
+  unitTestFilter: function(test){
     var origOptions = conn.zongji.options;
 
     conn.zongji.set({
@@ -65,7 +64,32 @@ module.exports = {
     // Restore original emitter
     delete conn.zongji.emit;
     conn.zongji.set(origOptions);
-    
+
     test.done();
+  },
+  integrationTestFilter: function(test){
+    // Set includeSchema to not include anything, recieve no row events
+    // Ensure that filters are applied
+    var origOptions = conn.zongji.options;
+    var testTable = 'filter_test';
+    var includeSchema = {};
+    // Uncomment the following line to manually test this test:
+    // includeSchema[settings.database] = [ testTable ];
+    conn.zongji.set({
+      includeEvents: ['tablemap', 'writerows', 'updaterows', 'deleterows'],
+      includeSchema: includeSchema
+    });
+    querySequence(conn.db, [
+      'DROP TABLE IF EXISTS ' + conn.escId(testTable),
+      'CREATE TABLE ' + conn.escId(testTable) + ' (col INT UNSIGNED)',
+      'INSERT INTO ' + conn.escId(testTable) + ' (col) VALUES (10)',
+      'UPDATE ' + conn.escId(testTable) + ' SET col = 15',
+      'DELETE FROM ' + conn.escId(testTable)
+    ], function(){
+      conn.zongji.set(origOptions);
+      test.equal(conn.eventLog.length, 0);
+      test.equal(conn.errorLog.length, 0);
+      test.done();
+    });
   }
 }