diff --git a/config/producer.yml b/config/producer.yml index 2e2c4b7..848e8a1 100644 --- a/config/producer.yml +++ b/config/producer.yml @@ -5,6 +5,7 @@ deleteNonEmpty: false amqp: amqp://user:password@localhost:5672 pingInterval: 60 flushInterval: 10 +serverId: 1 db: host: localhost port: 3306 diff --git a/mycdc.js b/mycdc.js index 9529672..64e93ed 100644 --- a/mycdc.js +++ b/mycdc.js @@ -60,7 +60,7 @@ module.exports = class MyCDC { tableInfo = { queues: new Map(), events: new Map(), - columns: false, + columnSet: false, fk: 'id' }; tableMap.set(tableName, tableInfo); @@ -79,13 +79,13 @@ module.exports = class MyCDC { const columns = table.columns; if (columns) { - if (tableInfo.columns === false) - tableInfo.columns = new Set(); - if (tableInfo.columns !== true) + if (tableInfo.columnSet === false) + tableInfo.columnSet = new Set(); + if (tableInfo.columnSet !== true) for (const column of columns) - tableInfo.columns.add(column); + tableInfo.columnSet.add(column); } else - tableInfo.columns = true; + tableInfo.columnSet = true; if (table.id) tableInfo.id = table.id; @@ -96,8 +96,15 @@ module.exports = class MyCDC { } const includeSchema = {}; - for (const [schemaName, tableMap] of this.schemaMap) + + for (const [schemaName, tableMap] of this.schemaMap) { includeSchema[schemaName] = Array.from(tableMap.keys()); + + for (const [tableName, tableInfo] of tableMap) { + if (tableInfo.columnSet !== true) + tableInfo.columns = Array.from(tableInfo.columnSet.keys()); + } + } this.opts = { includeEvents: [ @@ -107,7 +114,8 @@ module.exports = class MyCDC { 'updaterows', 'deleterows' ], - includeSchema + includeSchema, + serverId: conf.serverId }; if (conf.testMode) @@ -353,9 +361,9 @@ module.exports = class MyCDC { if (isUpdate) { rows = []; cols = new Set(); - let columns = tableInfo.columns === true + const columns = tableInfo.columnSet === true ? Object.keys(evt.rows[0].after) - : tableInfo.columns.keys(); + : tableInfo.columns; for (const row of evt.rows) { let nColsChanged = 0; diff --git a/package.json b/package.json index fa89ccd..42eda9c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "mycdc", - "version": "0.0.13", + "version": "0.0.14", "author": "Verdnatura Levante SL", "description": "Asynchronous DB calculations reading the binary log", "license": "GPL-3.0",