Multi-queue alpha, yml config, code clean, refactor

config.json

@@ -1,57 +0,0 @@
-{
-  "debug": true,
-  "testMode": false,
-  "db": {
-    "host": "localhost",
-    "port": 3306,
-    "user": "zongji",
-    "password": "password",
-    "database": "util"
-  },
-  "consumerDb": {
-    "host": "localhost",
-    "port": 3306,
-    "user": "zongji",
-    "password": "password",
-    "database": "util"
-  },
-  "amqp": "amqp://user:password@localhost:5672",
-  "includeEvents": [
-    "rotate",
-    "tablemap",
-    "writerows",
-    "updaterows",
-    "deleterows"
-  ],
-  "pingInterval": 60,
-  "flushInterval": 5000,
-  "queue": "orderRecalc",
-  "addQuery": "INSERT INTO `hedera`.`orderRecalc` (`orderFk`) VALUES ?",
-  "recalcQuery": "CALL `hedera`.`order_recalc`(?)",
-  "includeSchema": {
-    "hedera": {
-      "order": {
-        "fk": "id",
-        "events": ["updaterows"],
-        "columns": [
-          "id",
-          "address_id",
-          "company_id",
-          "date_send",
-          "customer_id"
-        ]
-      },
-      "orderRow": {
-        "fk": "orderFk",
-        "columns": [
-          "id",
-          "orderFk",
-          "itemFk",
-          "warehouseFk",
-          "shipment",
-          "amount"
-        ]
-      }
-    }
-  }
-}

config.yml

@@ -0,0 +1,66 @@
+
+debug: true
+testMode: true
+db:
+  host: localhost
+  port: 3306
+  user: zongji
+  password: password
+  database: util
+consumerDb: 
+  host: localhost
+  port: 3306
+  user: zongji
+  password: password
+  database: util
+amqp: amqp://user:password@localhost:5672
+pingInterval: 60
+flushInterval: 5000
+queues:
+  orderTotal:
+    query: CALL hedera.order_recalc(?)
+    mode: fk
+    includeSchema:
+      hedera:
+        order:
+          fk: id
+          events:
+          - updaterows
+          columns:
+          - id
+          - address_id
+          - company_id
+          - date_send
+          - customer_id
+        orderRow:
+          fk: orderFk
+          table: order
+          columns:
+          - id
+          - orderFk
+          - itemFk
+          - warehouseFk
+          - shipment
+          - amount
+  comparative: 
+    query: CALL vn.comparative_refresh(?table, ?id, ?data)
+    mode: changes
+    includeSchema:
+      vn: 
+        ticket:
+          id: id
+          events:
+          - updaterows
+          columns:
+          - id
+          - shipped
+          - warehouseFk
+          - isDeleted
+        sale:
+          id: id
+          columns:
+          - id
+          - ticketFk
+          - itemFk
+          - quantity
+          - price

consumer.js

@@ -1,26 +1,21 @@
-
+require('require-yaml');
+require('colors');
 const fs = require('fs');
 const path = require('path');
-const defaultConfig = require('./config.json');
-
 const mysql = require('mysql2/promise');
 const amqp = require('amqplib');
-require('colors');
 
-const config = Object.assign({}, defaultConfig);
+class Consumer {
-const localPath = path.join(__dirname, 'config.local.json');
+  async start() {
+    const defaultConfig = require('./config.yml');
+    const config = this.config = Object.assign({}, defaultConfig);
+    const localPath = path.join(__dirname, 'config.local.yml');
     if (fs.existsSync(localPath)) {
       const localConfig = require(localPath);
       Object.assign(config, localConfig);
     }
 
-class Consumer {
+    if (config.testMode)
-  constructor(config) {
-    this.config = config;
-  }
-
-  async start() {
-    if (this.config.testMode)
       console.log('Test mode enabled, just logging queries to console.');
 
     console.log('Starting process.');
@@ -35,20 +30,25 @@ class Consumer {
   }
 
   async init() {
+    const config = this.config;
     this.onErrorListener = err => this.onError(err);
 
-    this.db = await mysql.createConnection(this.config.consumerDb);
+    this.db = await mysql.createConnection(config.consumerDb);
     this.db.on('error', this.onErrorListener);
 
     this.pingInterval = setInterval(
-      () => this.connectionPing(), this.config.pingInterval * 1000);
+      () => this.connectionPing(), config.pingInterval * 1000);
 
-    this.consumer = await amqp.connect(this.config.amqp);
+    this.consumer = await amqp.connect(config.amqp);
     this.channel = await this.consumer.createChannel();
-    this.channel.assertQueue(this.config.queue, {
+
+    for (const queueName in config.queues) {
+      await this.channel.assertQueue(queueName, {
         durable: true
       });
-    this.channel.consume(this.config.queue, msg => this.onConsume(msg));
+      await this.channel.consume(queueName,
+        msg => this.onConsume(msg, queueName));
+    }
   }
 
   async end(silent) {
@@ -72,18 +72,28 @@ class Consumer {
     await this.db.ping();
   }
 
-  async onConsume(msg) {
+  async onConsume(msg, queueName) {
-    const fks = JSON.parse(msg.content.toString());
+    const config = this.config;
-    if (this.config.debug)
-      console.debug('RabbitMQ message'.blue, fks);
 
-    for (const fk of fks) {
+    const data = JSON.parse(msg.content.toString());
-      if (this.config.debug)
+    if (config.debug)
-        console.debug('Query'.blue, this.config.recalcQuery.yellow);
+      console.debug('Message:'.blue, queueName.yellow, fks);
-      await this.db.query(this.config.recalcQuery, fk);
+
+    const queue = config.queues[queueName];
+    const query = queue.query;
+    if (!query) return;
+
+    if (!config.testMode)
+    switch(queue.mode) {
+    case 'fk':
+      for (const fk of data.fks)
+          await this.db.query(query, fk);
+      break;
+    case 'changes':
+      break;
     }
 
-    this.channel.ack(msg);
+    await this.channel.ack(msg);
   }
 
   async tryRestart() {
@@ -118,10 +128,8 @@ class Consumer {
   }
 }
 
-let consumer;
-
 async function main() {
-  consumer = new Consumer(config)
+  const consumer = new Consumer()
   await consumer.start();
 
   process.on('SIGINT', async function() {

index.js

@@ -1,20 +1,7 @@
-
-const fs = require('fs');
-const path = require('path');
-const defaultConfig = require('./config.json');
 const MyCDC = require('./mycdc');
 
-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);
-}
-
-let mycdc;
-
 async function main() {
-  mycdc = new MyCDC(config)
+  const mycdc = new MyCDC()
   await mycdc.start();
 
   process.on('SIGINT', async function() {

mycdc.js

@@ -1,62 +1,107 @@
+require('require-yaml');
+require('colors');
+const fs = require('fs');
+const path = require('path');
 const ZongJi = require('./zongji');
 const mysql = require('mysql2/promise');
 const amqp = require('amqplib');
-require('colors');
 
-const allEvents = new Set([
+const allEvents = [
   'writerows',
   'updaterows',
   'deleterows'
-  ]);
+];
 
 module.exports = class MyCDC {
-  constructor(config) {
+  constructor() {
-    this.config = config;
     this.running = false;
     this.filename = null;
     this.position = null;
     this.schemaMap = new Map();
     this.fks = new Set();
-
+    this.queues = {};
-    const includeSchema = {};
-    for (const schemaName in this.config.includeSchema) {
-      const schema = this.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;
-      this.schemaMap.set(schemaName, tableMap);
-    }
-  
-    this.opts = {
-      includeEvents: this.config.includeEvents,
-      includeSchema
-    };
   }
 
   async start() {
-    if (this.config.testMode)
+    const defaultConfig = require('./config.yml');
+    const config = this.config = Object.assign({}, defaultConfig);
+    const localPath = path.join(__dirname, 'config.local.yml');
+    if (fs.existsSync(localPath)) {
+      const localConfig = require(localPath);
+      Object.assign(config, localConfig);
+    }
+
+    const queues = config.queues;
+    for (const queueName in queues) {
+      const includeSchema = queues[queueName].includeSchema;
+      for (const schemaName in includeSchema) {
+        let tableMap = this.schemaMap.get(schemaName);
+        if (!tableMap) {
+          tableMap = new Map();
+          this.schemaMap.set(schemaName, tableMap);
+        }
+
+        const schema = includeSchema[schemaName];
+        for (const tableName in schema) {
+          const table = schema[tableName];
+          //if (typeof table !== 'object') continue;
+
+          let tableInfo = tableMap.get(tableName);
+          if (!tableInfo) {
+            tableInfo = {
+              queues: new Map(),
+              events: new Map(),
+              columns: new Map(),
+              fk: 'id'
+            };
+            tableMap.set(tableName, tableInfo);
+          }
+          tableInfo.queues.set(queueName, table);
+
+          const events = table.events || allEvents;
+          for (const event of events) {
+            let eventInfo = tableInfo.events.get(event);
+            if (!eventInfo) {
+              eventInfo = [];
+              tableInfo.events.set(event, eventInfo);
+            }
+            eventInfo.push(queueName);
+          }
+
+          const columns = table.columns;
+          for (const column of columns) {
+            let columnInfo = tableInfo.columns.get(column);
+            if (!columnInfo) {
+              columnInfo = [];
+              tableInfo.columns.set(column, columnInfo);
+            }
+            columnInfo.push(queueName);
+          }
+
+          if (table.id)
+            tableInfo.id = table.id;
+        }
+
+        this.schemaMap.set(schemaName, tableMap);
+      }
+    }
+
+    const includeSchema = {};
+    for (const [schemaName, tableMap] of this.schemaMap)
+      includeSchema[schemaName] = Array.from(tableMap.keys());
+  
+    this.opts = {
+      includeEvents: [
+        'rotate',
+        'tablemap',
+        'writerows',
+        'updaterows',
+        'deleterows'
+      ],
+      includeSchema
+    };
+
+    if (config.testMode)
       console.log('Test mode enabled, just logging queries to console.');
 
     console.log('Starting process.');
@@ -71,25 +116,29 @@ module.exports = class MyCDC {
   }
 
   async init() {
-    this.debug('DbAsync', 'Initializing.');
+    const config = this.config;
+    this.debug('MyCDC', 'Initializing.');
     this.onErrorListener = err => this.onError(err);
 
     // DB connection
 
-    this.db = await mysql.createConnection(this.config.db);
+    this.db = await mysql.createConnection(config.db);
     this.db.on('error', this.onErrorListener);
 
     // RabbitMQ
 
-    this.publisher = await amqp.connect(this.config.amqp);
+    this.publisher = await amqp.connect(config.amqp);
     this.channel = await this.publisher.createChannel();
-    this.channel.assertQueue(this.config.queue, {
+
+    for (const queueName in config.queues) {
+      await this.channel.assertQueue(queueName, {
         durable: true
       });
+    }
 
     // Zongji
 
-    const zongji = new ZongJi(this.config.db);
+    const zongji = new ZongJi(config.db);
     this.zongji = zongji;
 
     this.onBinlogListener = evt => this.onBinlog(evt);
@@ -97,7 +146,7 @@ module.exports = class MyCDC {
 
     const [res] = await this.db.query(
       'SELECT `logName`, `position` FROM `binlogQueue` WHERE code = ?',
-      [this.config.queue]
+      [config.queue]
     );
     if (res.length) {
       const [row] = res;
@@ -132,21 +181,21 @@ module.exports = class MyCDC {
     this.zongji.on('error', this.onErrorListener);
 
     this.flushInterval = setInterval(
-      () => this.flushQueue(), this.config.flushInterval);
+      () => this.flushQueue(), config.flushInterval);
     this.pingInterval = setInterval(
-      () => this.connectionPing(), this.config.pingInterval * 1000);
+      () => this.connectionPing(), config.pingInterval * 1000);
 
     // Summary
 
     this.running = true;
-    this.debug('DbAsync', 'Initialized.');
+    this.debug('MyCDC', 'Initialized.');
   }
 
   async end(silent) {
     const zongji = this.zongji;
     if (!zongji) return;
 
-    this.debug('DbAsync', 'Ending.');
+    this.debug('MyCDC', 'Ending.');
 
     // Zongji
 
@@ -194,7 +243,7 @@ module.exports = class MyCDC {
 
     // Summary
 
-    this.debug('DbAsync', 'Ended.');
+    this.debug('MyCDC', 'Ended.');
   }
 
   async tryRestart() {
@@ -242,56 +291,107 @@ module.exports = class MyCDC {
     const tableInfo = tableMap.get(table.tableName);
     if (!tableInfo) return;
 
-    if (!tableInfo.events.has(eventName)) return;
+    const queueNames = tableInfo.events.get(eventName);
+    if (!queueNames) return;
 
-    let column;
     const rows = evt.rows;
+    const queues = this.config.queues;
+    const tableQueues = tableInfo.queues;
 
-    const fks = new Set();
+    const changes = new Map();
+    for (const queueName of queueNames) {
+      const change = {
+        mode: queues[queueName].mode
+      };
+      changes.set(queueName, change);
 
-    if (eventName === 'updaterows') {
+      switch(change.mode)  {
-      if (tableInfo.columns !== true) {
+      case 'fk':
-        let changes = false;
+        change.fks = new Set();
-        for (const row of rows) {
+        break;
-          const after = row.after;
+      case 'changes':
-          for (const col in after) {
+        change.rows = {};
-            if (tableInfo.columns.has(col) && !equals(after[col], row.before[col])) {
+        break;
-              fks.add(after[tableInfo.fk]);
+      }
-              changes = true;
+    }
-              if (!column) column = col;
+
+    function addChange(row, queueNames) {
+      for (const queueName of queueNames) {
+        const queueInfo = tableQueues.get(queueName);
+        const change = changes.get(queueName);
+
+        switch(change.mode)  {
+        case 'fk':
+          change.fks.add(row[queueInfo.fk]);
+          break;
+        case 'changes':
+          const queueRow = {};
+          for (const column of queueInfo.columns)
+            if (row[column] !== undefined)
+              queueRow[column] = row[column];
+          change.rows[row[queueInfo.id]] = queueRow;
           break;
         }
       }
     }
+
+    const columnMap = tableInfo.columns;
+    const columns = columnMap.keys();
+
+    if (eventName === 'updaterows') {
+      const changedQueues = new Set();
+      for (const row of rows) {
+        changedQueues.clear();
+        const after = row.after;
+
+        for (const col of columns) {
+          if (after[col] === undefined
+          || equals(after[col], row.before[col]))
+            continue;
+
+          for (const queue of columnMap.get(col))
+            changedQueues.add(queue);
+
+          if (changedQueues.size === queueNames.length)
+            break;
+        }
+
+        if (changedQueues.size)
+          addChange(after, changedQueues);
+      }
       if (!changes) return;
     } else {
       for (const row of rows)
-          fks.add(row.after[tableInfo.fk]);
+        addChange(row, queueNames);
-      }
-    } else {
-      for (const row of rows)
-        fks.add(row[tableInfo.fk]);
     }
 
-    if (fks.size) {
+    for (const [queueName, change] of changes) {
-      const data = JSON.stringify(Array.from(fks));
+      const jsonData = {
-      this.channel.sendToQueue(this.config.queue,
+        eventName,
-        Buffer.from(data));
+        table: table.tableName,
-      this.debug('Queued', data);
+        schema: table.parentSchema,
+        mode: change.mode
+      };
+
+      let nChanges;
+      switch(change.mode)  {
+      case 'fk':
+        jsonData.fks = Array.from(change.fks);
+        nChanges = change.fks.size;
+        break;
+      case 'changes':
+        jsonData.rows = change.rows;
+        nChanges = change.rows.length;
+        break;
       }
 
-    const row = eventName === 'updaterows'
+      if (!nChanges) continue;
-      ? rows[0].after
-      : rows[0];
 
-    if (this.config.debug) {
+      const data = JSON.stringify(jsonData);
-      console.debug(`[${eventName}] ${table.tableName}: ${rows.length}`);
+      this.channel.sendToQueue(queueName,
-      console.debug(`  ${tableInfo.fk}: ${row[tableInfo.fk]}`);
+        Buffer.from(data), {persistent: true});
-      if (column) {
+
-        let before = formatValue(rows[0].before[column]);
+      console.debug('Queued'.blue, queueName.yellow, `[${eventName}] ${table.tableName}`);
-        let after = formatValue(rows[0].after[column]);
-        console.debug(`  ${column}: ${before} <- ${after}`);
-      }
     }
 
     this.position = evt.nextPosition;

package-lock.json

@@ -1,5 +1,5 @@
 {
-  "name": "db-async",
+  "name": "mycdc",
   "lockfileVersion": 2,
   "requires": true,
   "packages": {
@@ -8,6 +8,7 @@
         "amqplib": "^0.10.3",
         "colors": "^1.4.0",
         "mysql2": "^2.3.3",
+        "require-yaml": "^0.0.1",
         "zongji": "file:../zongji"
       }
     },
@@ -39,6 +40,11 @@
         "node": ">=10"
       }
     },
+    "node_modules/argparse": {
+      "version": "2.0.1",
+      "resolved": "https://registry.npmjs.org/argparse/-/argparse-2.0.1.tgz",
+      "integrity": "sha512-8+9WqebbFzpX9OR+Wa6O29asIogeRMzcGtAINdpMHHyAg10f05aSFVBbcEqGf/PXw1EjAZ+q2/bEBg3DvurK3Q=="
+    },
     "node_modules/buffer-more-ints": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
@@ -115,6 +121,17 @@
       "resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
       "integrity": "sha512-D2S+3GLxWH+uhrNEcoh/fnmYeP8E8/zHl644d/jdA0g2uyXvy3sb0qxotE+ne0LtccHknQzWwZEzhak7oJ0COQ=="
     },
+    "node_modules/js-yaml": {
+      "version": "4.1.0",
+      "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.1.0.tgz",
+      "integrity": "sha512-wpxZs9NoxZaJESJGIZTyDEaYpl0FKSA+FB9aJiyemKhMwkxQg63h4T1KJgUGHpTqPDNRcmmYLugrRjJlBtWvRA==",
+      "dependencies": {
+        "argparse": "^2.0.1"
+      },
+      "bin": {
+        "js-yaml": "bin/js-yaml.js"
+      }
+    },
     "node_modules/long": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/long/-/long-4.0.0.tgz",
@@ -200,6 +217,14 @@
         "string_decoder": "~0.10.x"
       }
     },
+    "node_modules/require-yaml": {
+      "version": "0.0.1",
+      "resolved": "https://registry.npmjs.org/require-yaml/-/require-yaml-0.0.1.tgz",
+      "integrity": "sha512-M6eVEgLPRbeOhgSCnOTtdrOOEQzbXRchg24Xa13c39dMuraFKdI9emUo97Rih0YEFzSICmSKg8w4RQp+rd9pOQ==",
+      "dependencies": {
+        "js-yaml": ""
+      }
+    },
     "node_modules/requires-port": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
@@ -274,6 +299,11 @@
         "url-parse": "~1.5.10"
       }
     },
+    "argparse": {
+      "version": "2.0.1",
+      "resolved": "https://registry.npmjs.org/argparse/-/argparse-2.0.1.tgz",
+      "integrity": "sha512-8+9WqebbFzpX9OR+Wa6O29asIogeRMzcGtAINdpMHHyAg10f05aSFVBbcEqGf/PXw1EjAZ+q2/bEBg3DvurK3Q=="
+    },
     "buffer-more-ints": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
@@ -333,6 +363,14 @@
       "resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
       "integrity": "sha512-D2S+3GLxWH+uhrNEcoh/fnmYeP8E8/zHl644d/jdA0g2uyXvy3sb0qxotE+ne0LtccHknQzWwZEzhak7oJ0COQ=="
     },
+    "js-yaml": {
+      "version": "4.1.0",
+      "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.1.0.tgz",
+      "integrity": "sha512-wpxZs9NoxZaJESJGIZTyDEaYpl0FKSA+FB9aJiyemKhMwkxQg63h4T1KJgUGHpTqPDNRcmmYLugrRjJlBtWvRA==",
+      "requires": {
+        "argparse": "^2.0.1"
+      }
+    },
     "long": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/long/-/long-4.0.0.tgz",
@@ -411,6 +449,14 @@
         "string_decoder": "~0.10.x"
       }
     },
+    "require-yaml": {
+      "version": "0.0.1",
+      "resolved": "https://registry.npmjs.org/require-yaml/-/require-yaml-0.0.1.tgz",
+      "integrity": "sha512-M6eVEgLPRbeOhgSCnOTtdrOOEQzbXRchg24Xa13c39dMuraFKdI9emUo97Rih0YEFzSICmSKg8w4RQp+rd9pOQ==",
+      "requires": {
+        "js-yaml": ""
+      }
+    },
     "requires-port": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",

package.json

@@ -3,6 +3,7 @@
     "amqplib": "^0.10.3",
     "colors": "^1.4.0",
     "mysql2": "^2.3.3",
+    "require-yaml": "^0.0.1",
     "zongji": "file:../zongji"
   }
 }

zongji.sql

@@ -10,5 +10,3 @@ CREATE USER 'zongji'@'%' IDENTIFIED BY 'password';
 GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT ON *.* TO 'zongji'@'%';
 
 GRANT INSERT, DELETE ON `util`.* TO 'zongji'@'%';
-GRANT INSERT ON `hedera`.`orderRecalc` TO 'zongji'@'%';
-GRANT INSERT ON `vn`.`ticketRecalc` TO 'zongji'@'%';