mycdc/index.js

const ZongJi = require('./zongji');
const mysql = require('mysql2/promise');
const fs = require('fs');
const path = require('path');
const defaultConfig = require('./config.json');

const config = Object.assign({}, defaultConfig);
const localPath = path.join(__dirname, 'config.local.json');
if (fs.existsSync(localPath)) {
  const localConfig = require(localPath);
  Object.assign(config, localConfig);
}

const zongji = new ZongJi(config.db);
const schemaMap = new Map();

const allEvents = new Set([
  'writerows',
  'updaterows',
  'deleterows'
]);

let db;
let nextPosition;
let filename;
const fks = new Set();

async function main() {
  db = await mysql.createConnection(config.db);

  const includeSchema = {};
  for (const schemaName in config.includeSchema) {
    const schema = config.includeSchema[schemaName];
    const tables = [];
    const tableMap = new Map();

    for (const tableName in schema) {
      const table = schema[tableName];
      tables.push(tableName);

      const tableInfo = {
        events: allEvents,
        columns: true,
        fk: 'id'
      };
      tableMap.set(tableName, tableInfo);
  
      if (typeof table === 'object') {
        if (Array.isArray(table.events))
          tableInfo.events = new Set(table.events);
        if (Array.isArray(table.columns))
          tableInfo.columns = new Set(table.columns);
        if (table.fk)
          tableInfo.fk = table.fk;
      }
    }

    includeSchema[schemaName] = tables;
    schemaMap.set(schemaName, tableMap);
  }

  const opts = {
    includeEvents: config.includeEvents,
    includeSchema
  };

  const [res] = await db.query(
    'SELECT `logName`, `position` FROM `binlogQueue` WHERE code = ?',
    [config.queue]
  );
  if (res.length) {
    const [row] = res;
    filename = row.logName;
    position = row.position;
    Object.assign(opts, {filename, position});
  } else
    opts.startAtEnd = true;

  zongji.start(opts);
  setInterval(flushQueue, config.flushInterval);
  console.log('Listenig binary log events.');
  
  process.on('SIGINT', async function() {
    console.log('Got SIGINT.');
    zongji.stop();
    await db.end();
    process.exit();
  });
}

async function flushQueue() {
  console.log('==========================================================')
  console.log('Flush:', `filename=${filename}`, `position=${nextPosition}`);
  console.log(fks);
  if (!fks.size) return;

  const ids = [];
  for (const fk of fks) ids.push([fk]);
  await db.query(config.addQuery, [ids]);
  await db.query(
    'REPLACE INTO `binlogQueue` (`code`, `logName`, `position`) VALUES (?, ?, ?)',
    [config.queue, filename, nextPosition]
  );
  fks.clear();
}

function equals(a, b) {
  if (a === b)
    return true;
  const type = typeof a;
  if (a == null || b == null || type !== typeof b)
    return false;
  if (type === 'object' && a.constructor === b.constructor) {
    if (a instanceof Date)
      return a.getTime() === b.getTime();
  }
  return false;
}

zongji.on('binlog', function(evt) {
  //evt.dump();
  const eventName = evt.getEventName();
  const table = evt.tableMap[evt.tableId];
  if (eventName === 'tablemap') return;

  const tableMap = schemaMap.get(table.parentSchema);
  if (!tableMap) return;

  const tableInfo = tableMap.get(table.tableName);
  if (!tableInfo) return;

  if (!tableInfo.events.has(eventName)) return;

  let column;
  const rows = evt.rows;

  if (eventName === 'updaterows') {
    if (tableInfo.columns !== true) {
      let changes = false;
      for (const row of rows) {
        const after = row.after;
        for (const col in after) {
          if (tableInfo.columns.has(col) && !equals(after[col], row.before[col])) {
            fks.add(after[tableInfo.fk]);
            changes = true;
            if (!column) column = col;
            break;
          }
        }
      }
      if (!changes) return;
    } else {
      for (const row of rows)
        fks.add(row.after[tableInfo.fk]);
    }
  } else {
    for (const row of rows)
      fks.add(row[tableInfo.fk]);
  }

  const row = eventName === 'updaterows'
    ? rows[0].after
    : rows[0];

  console.log(`[${eventName}] ${table.tableName}: ${rows.length}`);
  console.log(`  ${tableInfo.fk}: ${row[tableInfo.fk]}`);
  if (column)
    console.log(`  ${column}: ${rows[0].after[column]} <- ${rows[0].before[column]}`);

  nextPosition = evt.nextPosition;
  filename = zongji.options.filename;
});

main();