feat: refs #4685 AMQ queue prefix, stock queue remove isDelivered

config/consumer.yml

@@ -4,6 +4,7 @@ defaults:
   mode: fk
   flushInterval: 5000
   amqpPrefetch: 100
+amqPrefix: cdc
 amqp: amqp://user:password@localhost:5672
 db: 
   host: localhost

config/producer.yml

@@ -2,6 +2,7 @@ code: mycdc
 debug: false
 testMode: false
 deleteNonEmpty: false
+amqPrefix: cdc
 amqp: amqp://user:password@localhost:5672
 pingInterval: 60
 flushInterval: 10

lib/queue.js

@@ -6,12 +6,15 @@ module.exports = class Queue {
     async consume() {
       const channel = await this.consumer.amqpConn.createChannel();
       channel.prefetch(this.conf.amqpPrefetch);
-      await channel.assertQueue(this.name, {
+      const {amqPrefix} = this.consumer.conf;
+      const amqQueue = `${amqPrefix}.${this.name}`;
+
+      await channel.assertQueue(amqQueue, {
         durable: true
       });
       this.channel = channel;
   
-      await channel.consume(this.name,
+      await channel.consume(amqQueue,
         msg => this.onConsume(msg));
     }
 }

mycdc.js

@@ -147,17 +147,19 @@ module.exports = class MyCDC {
     this.publisher = await amqp.connect(conf.amqp);
     const channel = this.channel = await this.publisher.createChannel();
 
+    const {amqPrefix} = conf;
     for (const tableMap of this.schemaMap.values()) {
       for (const tableName of tableMap.keys()) {
-        await channel.assertExchange(tableName, 'headers', {
+        await channel.assertExchange(`${amqPrefix}.${tableName}`, 'headers', {
           durable: true
         });
       }
     }
     for (const queueName in this.queuesConf) {
+      const amqQueue = `${amqPrefix}.${queueName}`;
       const options = conf.deleteNonEmpty ? {} : {ifEmpty: true};
-      await channel.deleteQueue(queueName, {options});
+      await channel.deleteQueue(amqQueue, {options});
-      await channel.assertQueue(queueName, {
+      await channel.assertQueue(amqQueue, {
         durable: true
       });
 
@@ -167,15 +169,15 @@ module.exports = class MyCDC {
         for (const tableName in schema) {
           const table = schema[tableName];
           const events = table.events || allEvents;
+          let args = {'x-match': 'any'};
           for (const event of events) {
-            let args;
+            if (event === 'updaterows' && table.columns)
-            if (event === 'updaterows' && table.columns) {
-              args = {'x-match': 'any'};
               table.columns.map(c => args[c] = true);
-            } else
+            else
-              args = {'z-event': event};
+              args[`z-${event}`] = true;
-            await channel.bindQueue(queueName, tableName, '', args);
           }
+          await channel.bindQueue(amqQueue,
+            `${amqPrefix}.${tableName}`, '', args);
         }
       }
     }
@@ -392,7 +394,7 @@ module.exports = class MyCDC {
     };
 
     let headers = {};
-    headers['z-event'] = eventName;
+    headers[`z-${eventName}`] = true;
     if (isUpdate) {
       for (const col of cols)
         headers[col] = true;
@@ -404,8 +406,9 @@ module.exports = class MyCDC {
       headers
     };
 
+    const {amqPrefix} = this.conf;
     const jsonData = JSON.stringify(data);
-    this.channel.publish(tableName, '',
+    this.channel.publish(`${amqPrefix}.${tableName}`, '',
       Buffer.from(jsonData), options);
 
     if (this.debug) {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "mycdc",
-  "version": "0.0.27",
+  "version": "0.0.28",
   "author": "Verdnatura Levante SL",
   "description": "Asynchronous DB calculations reading the binary log",
   "license": "GPL-3.0",

queues/stock.yml

@@ -16,7 +16,6 @@ includeSchema:
       - shipped
       - warehouseInFk
       - warehouseOutFk
-      - isDelivered
       - isReceived
       events:
       - updaterows
@@ -55,7 +54,6 @@ includeSchema:
       - itemFk
       - quantity
       - created
-      - isPicked
   hedera:
     order:
       key: id