index.js

const mysql = require('@vlasky/mysql');
const util = require('util');
const EventEmitter = require('events').EventEmitter;
const initBinlogClass = require('./lib/sequence/binlog');

const ConnectionConfigMap = {
  'Connection': obj => obj.config,
  'Pool': obj => obj.config.connectionConfig,
};

const TableInfoQueryTemplate = `SELECT 
  COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME, 
  COLUMN_COMMENT, COLUMN_TYPE 
  FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='%s' 
  ORDER BY ORDINAL_POSITION`;

function ZongJi(dsn) {
  EventEmitter.call(this);

  this._options({});
  this._filters({});
  this.ctrlCallbacks = [];
  this.tableMap = {};
  this.ready = false;
  this.useChecksum = false;

  this._establishConnection(dsn);
}

util.inherits(ZongJi, EventEmitter);

// dsn - can be one instance of Connection or Pool / object / url string
ZongJi.prototype._establishConnection = function(dsn) {
  const createConnection = (options) => {
    let connection = mysql.createConnection(options);
    connection.on('error', this.emit.bind(this, 'error'));
    connection.on('unhandledError', this.emit.bind(this, 'error'));
    // don't need to call connection.connect() here
    // we use implicitly established connection
    // see https://github.com/mysqljs/mysql#establishing-connections
    return connection;
  };

  const configFunc = ConnectionConfigMap[dsn.constructor.name];
  let binlogDsn;

  if (typeof dsn === 'object' && configFunc) {
    // dsn is a pool or connection object
    let conn = dsn; // reuse as ctrlConnection
    this.ctrlConnection = conn;
    this.ctrlConnectionOwner = false;
    binlogDsn = Object.assign({}, configFunc(conn));
  }

  if (!binlogDsn) {
    // assuming that the object passed is the connection settings
    this.ctrlConnectionOwner = true;
    this.ctrlConnection = createConnection(dsn);
    binlogDsn = dsn;
  }

  this.connection = createConnection(binlogDsn);
};

ZongJi.prototype._isChecksumEnabled = function(next) {
  const SelectChecksumParamSql = 'select @@GLOBAL.binlog_checksum as checksum';
  const SetChecksumSql = 'set @master_binlog_checksum=@@global.binlog_checksum';

  const query = (conn, sql) => {
    return new Promise(
      (resolve, reject) => {
        conn.query(sql, (err, result) => {
          if (err) {
            reject(err);
          }
          else {
            resolve(result);
          }
        });
      }
    );
  };

  let checksumEnabled = true;

  query(this.ctrlConnection, SelectChecksumParamSql)
    .then(rows => {
      if (rows[0].checksum === 'NONE') {
        checksumEnabled = false;
        return query(this.connection, 'SELECT 1');
      }

      if (checksumEnabled) {
        return query(this.connection, SetChecksumSql);
      }
    })
    .catch(err => {
      if (err.toString().match(/ER_UNKNOWN_SYSTEM_VARIABLE/)) {
        checksumEnabled = false;
        // a simple query to open this.connection
        return query(this.connection, 'SELECT 1');
      }
      else {
        next(err);
      }
    })
    .then(() => {
      next(null, checksumEnabled);
    });
};

ZongJi.prototype._findBinlogEnd = function(next) {
  this.ctrlConnection.query('SHOW BINARY LOGS', (err, rows) => {
    if (err) {
      // Errors should be emitted
      next(err);
    }
    else {
      next(null, rows.length > 0 ? rows[rows.length - 1] : null);
    }
  });
};

ZongJi.prototype._fetchTableInfo = function(tableMapEvent, next) {
  const sql = util.format(TableInfoQueryTemplate,
    tableMapEvent.schemaName, tableMapEvent.tableName);

  this.ctrlConnection.query(sql, (err, rows) => {
    if (err) {
      // Errors should be emitted
      this.emit('error', err);
      // This is a fatal error, no additional binlog events will be
      // processed since next() will never be called
      return;
    }

    if (rows.length === 0) {
      this.emit('error', new Error(
        'Insufficient permissions to access: ' +
        tableMapEvent.schemaName + '.' + tableMapEvent.tableName));
      // This is a fatal error, no additional binlog events will be
      // processed since next() will never be called
      return;
    }

    this.tableMap[tableMapEvent.tableId] = {
      columnSchemas: rows,
      parentSchema: tableMapEvent.schemaName,
      tableName: tableMapEvent.tableName
    };

    next();
  });
};

// #_options will reset all the options.
ZongJi.prototype._options = function({
  serverId,
  filename,
  position,
  startAtEnd,
}) {
  this.options = {
    serverId,
    filename,
    position,
    startAtEnd,
  };
};

// #_filters will reset all the filters.
ZongJi.prototype._filters = function({
  includeEvents,
  excludeEvents,
  includeSchema,
  excludeSchema,
}) {
  this.filters = {
    includeEvents,
    excludeEvents,
    includeSchema,
    excludeSchema,
  };
};

ZongJi.prototype.get = function(name) {
  let result;
  if (typeof name === 'string') {
    result = this.options[name];
  }
  else if (Array.isArray(name)) {
    result = name.reduce(
      (acc, cur) => {
        acc[cur] = this.options[cur];
        return acc;
      },
      {}
    );
  }

  return result;
};

// @options contains a list options
// - `serverId` unique identifier
// - `filename`, `position` the position of binlog to beigin with
// - `startAtEnd` if true, will update filename / postion automatically
// - `includeEvents`, `excludeEvents`, `includeSchema`, `exludeSchema` filter different binlog events bubbling
ZongJi.prototype.start = function(options = {}) {

  this._options(options);
  this._filters(options);

  const testChecksum = (resolve, reject) => {
    this._isChecksumEnabled((err, checksumEnabled) => {
      if (err) {
        reject(err);
      }
      else {
        this.useChecksum = checksumEnabled;
        resolve();
      }
    });
  };


  const findBinlogEnd = (resolve, reject) => {
    this._findBinlogEnd((err, result) => {
      if (err) {
        return reject(err);
      }

      if (result) {
        this._options(
          Object.assign({}, options, {
            filename: result.Log_name,
            position: result.File_size,
          })
        );
      }

      resolve();
    });
  };

  const binlogHandler = (error, event) => {
    if (error) {
      return this.emit('error', error);
    }

    // Do not emit events that have been filtered out
    if (event === undefined || event._filtered === true) return;

    switch (event.getTypeName()) {
      case 'TableMap': {
        const tableMap = this.tableMap[event.tableId];
        if (!tableMap || tableMap.tableName !== event.tableName || tableMap.columns.length !== event.columnCount) {
          this.connection.pause();
          this._fetchTableInfo(event, () => {
            // merge the column info with metadata
            event.updateColumnInfo();
            this.emit('binlog', event);
            this.connection.resume();
          });
          return;
        }
        break;
      }
      case 'Rotate':
        if (this.options.filename !== event.binlogName) {
          this.options.filename = event.binlogName;
        }
        break;
    }
    this.options.position = event.nextPosition;
    this.emit('binlog', event);
  };

  let promises = [new Promise(testChecksum)];

  if (this.options.startAtEnd) {
    promises.push(new Promise(findBinlogEnd));
  }

  Promise.all(promises)
    .then(() => {
      this.BinlogClass = initBinlogClass(this);
      this.ready = true;
      this.emit('ready');

      this.connection._protocol._enqueue(
        new this.BinlogClass(binlogHandler)
      );
    })
    .catch(err => {
      this.emit('error', err);
    });

};

ZongJi.prototype.stop = function() {
  // Binary log connection does not end with destroy()
  this.connection.destroy();
  this.ctrlConnection.query(
    'KILL ' + this.connection.threadId,
    () => {
      if (this.ctrlConnectionOwner) {
        this.ctrlConnection.destroy();
      }
      this.emit('stopped');
    }
  );
};

// It includes every events by default.
ZongJi.prototype._skipEvent = function(name) {
  const includes = this.filters.includeEvents;
  const excludes = this.filters.excludeEvents;

  let included = (includes === undefined) ||
    (Array.isArray(includes) && (includes.indexOf(name) > -1));
  let excluded = Array.isArray(excludes) && (excludes.indexOf(name) > -1);

  return excluded || !included;
};

// It doesn't skip any schema by default.
ZongJi.prototype._skipSchema = function(database, table) {
  const includes = this.filters.includeSchema;
  const excludes = this.filters.excludeSchema || {};

  let included = (includes === undefined) ||
    (
      (database in includes) &&
      (
        includes[database] === true ||
        (
          Array.isArray(includes[database]) &&
          includes[database].indexOf(table) > -1
        )
      )
    );
  let excluded = (database in excludes) &&
    (
      excludes[database] === true ||
      (
        Array.isArray(excludes[database]) &&
        excludes[database].indexOf(table) > -1
      )
    );

  return excluded || !included;
};

module.exports = ZongJi;