Save msg object when calling robot

Author: skadefro

OpenFlowNodeRED/src/amqp_publisher.ts

@@ -47,10 +47,10 @@ export class amqp_publisher {
         if (this.channel != null && this.channel != undefined) { await this.channel.close(); this.channel = null; }
         if (this.conn != null && this.conn != undefined) { await this.conn.close(); this.conn = null; }
     }
-    SendMessage(msg: string, queue: string): void {
-        var corr: string = this.generateUuid();
+    SendMessage(msg: string, queue: string, correlationId: string): void {
+        if (correlationId == null || correlationId == "") { correlationId = this.generateUuid(); }
         this._logger.info("SendMessage " + msg);
-        this.channel.sendToQueue(queue, Buffer.from(msg), { correlationId: corr, replyTo: this._ok.queue });
+        this.channel.sendToQueue(queue, Buffer.from(msg), { correlationId: correlationId, replyTo: this._ok.queue });
     }
     private _OnMessage(sender: amqp_publisher, msg: amqplib.ConsumeMessage): void {
         try {

OpenFlowNodeRED/src/nodered/nodes/amqp_nodes.ts

@@ -203,7 +203,7 @@ export class amqp_publisher_node {
             data.payload = msg.payload;
             data.jwt = msg.jwt;
             data._id = msg._id;
-            this.con.SendMessage(JSON.stringify(data), this.config.queue);
+            this.con.SendMessage(JSON.stringify(data), this.config.queue, null);
             this.node.status({});
         } catch (error) {
             NoderedUtil.HandleError(this, error);

OpenFlowNodeRED/src/nodered/nodes/rpa_nodes.ts

@@ -113,6 +113,13 @@ export class rpa_workflow_node {
             var json: string = msg.content.toString();
             var data = JSON.parse(json);
             result.jwt = data.jwt;
+            var correlationId = msg.properties.correlationId;
+            if (correlationId != null && this.messages[correlationId] != null) {
+                result = this.messages[correlationId];
+                if (data.payload.command == "invokecompleted" || data.payload.command == "invokefailed" || data.payload.command == "invokeaborted" || data.payload.command == "error") {
+                    delete this.messages[correlationId];
+                }
+            }
 
             if (data.payload.command == "invokecompleted") {
                 result.payload = data.payload.data;
@@ -131,9 +138,15 @@ export class rpa_workflow_node {
             NoderedUtil.HandleError(this, error);
         }
     }
+    messages: any[] = [];
     async oninput(msg: any) {
         try {
             this.node.status({});
+            var correlationId = Math.random().toString(36).substr(2, 9);
+            this.messages[correlationId] = msg;
+            if (msg.payload == null || typeof msg.payload == "string" || typeof msg.payload == "number") {
+                msg.payload = { "data": msg.payload };
+            }
             var rpacommand = {
                 command: "invoke",
                 workflowid: this.config.workflow,
@@ -143,7 +156,7 @@ export class rpa_workflow_node {
                 jwt: msg.jwt,
                 payload: rpacommand
             }
-            this.con.SendMessage(JSON.stringify(data), this.config.queue);
+            this.con.SendMessage(JSON.stringify(data), this.config.queue, correlationId);
             // var data: any = {};
             // data.payload = msg.payload;
             // data.jwt = msg.jwt;

VERSION

@@ -1 +1 @@
-0.0.321
+0.0.322