fix: refs #4409 Columns changed detection fix

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

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,9 +96,16 @@ 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: [
         'rotate',
@@ -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;

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",