Preserve function payloads with ampq pub

Author: skadefro

OpenFlowNodeRED/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/nodered",
-  "version": "1.2.56",
+  "version": "1.2.57",
   "description": "Simple wrapper around NodeRed, RabbitMQ and MongoDB to support a more scaleable NodeRed implementation.\r Also the \"backend\" for [OpenRPA](https://github.com/skadefro/OpenRPA)",
   "main": "index.js",
   "scripts": {

OpenFlowNodeRED/src/index.ts

@@ -108,19 +108,7 @@ function handle(signal, value) {
     }
 }
 Object.keys(signals).forEach((signal) => process.on(signal, handle));
-// process.on('SIGTERM', handle);
-// process.on('SIGINT', handle);
-// process.on('SIGUSR1', handle);
-// process.on('SIGPIPE', handle);
-// process.on('SIGHUP', handle);
-// process.on('SIGBREAK', handle);
-// process.on('SIGKILL', handle);
-// process.on('SIGWINCH', handle);
-// process.on('SIGSTOP', handle);
-// process.on('SIGBUS', handle);
-// process.on('SIGFPE', handle);
-// process.on('SIGSEGV', handle);
-// process.on('SIGILL', handle);
+
 
 let server: http.Server = null;
 (async function (): Promise<void> {
@@ -146,8 +134,6 @@ let server: http.Server = null;
         });
         socket.events.on("onopen", async () => {
             try {
-                // q.clientagent = "nodered";
-                // q.clientversion = Config.version;
                 let jwt: string = "";
                 if (Config.jwt !== "") {
                     jwt = Config.jwt;

OpenFlowNodeRED/src/nodered/nodes/amqp_nodes.ts

@@ -178,6 +178,7 @@ export class amqp_publisher_node {
     private connection: amqp_connection;
     private _onsignedin: any = null;
     private _onsocketclose: any = null;
+    private payloads: any = {};
     constructor(public config: Iamqp_publisher_node) {
         RED.nodes.createNode(this, config);
         try {
@@ -232,7 +233,11 @@ export class amqp_publisher_node {
         try {
             const result: any = {};
             result.amqpacknowledgment = ack;
-            const data = msg.data;
+            let data = msg.data;
+            if (!NoderedUtil.IsNullEmpty(data._msgid)) {
+                data = Object.assign(this.payloads[data._msgid], data);
+                delete this.payloads[data._msgid];
+            }
             result.payload = data.payload;
             result.jwt = data.jwt;
             if (data.command == "timeout") {
@@ -253,11 +258,13 @@ export class amqp_publisher_node {
             data.payload = msg.payload;
             data.jwt = msg.jwt;
             data._id = msg._id;
+            data._msgid = msg._msgid;
             const expiration: number = (typeof msg.expiration == 'number' ? msg.expiration : Config.amqp_message_ttl);
             const queue = this.config.queue;
             this.node.status({ fill: "blue", shape: "dot", text: "Sending message ..." });
             try {
                 await NoderedUtil.QueueMessage(this.websocket(), queue, this.localqueue, data, null, expiration);
+                this.payloads[msg._msgid] = msg;
             } catch (error) {
                 data.error = error;
                 this.node.send([null, data]);

VERSION

@@ -1 +1 @@
-1.2.56
+1.2.57

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/openflow",
-  "version": "1.2.56",
+  "version": "1.2.57",
   "description": "Simple wrapper around NodeRed, RabbitMQ and MongoDB to support a more scaleable NodeRed implementation.\r Also the \"backend\" for [OpenRPA](https://github.com/skadefro/OpenRPA)",
   "main": "index.js",
   "scripts": {