111

Author: skadefro

OpenFlow/src/WebSocketClient.ts

@@ -130,7 +130,7 @@ export class WebSocketClient {
         }
         var AssertQueueOptions: any = new Object(amqpwrapper.Instance().AssertQueueOptions);
         AssertQueueOptions.autoDelete = autoDelete;
-        var queuename = await amqpwrapper.Instance().AddQueueConsumer(queuename, AssertQueueOptions, this.jwt, async (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
+        var queuename = await amqpwrapper.Instance().AddQueueConsumer(this, queuename, AssertQueueOptions, this.jwt, async (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
             var _data = msg;
             try {
                 _data = await this.Queue(msg, queuename, options);

OpenFlow/src/WebSocketServer.ts

@@ -62,7 +62,7 @@ export class WebSocketServer {
             }
             var now = new Date();
             var seconds = (now.getTime() - cli.lastheartbeat.getTime()) / 1000;
-            if (seconds >= 20) {
+            if (seconds >= 60) {
                 if (cli.user != null) {
                     WebSocketServer._logger.info("client " + cli.user.name + "/" + cli.clientagent + " timeout, close down");
                 } else {

OpenFlow/src/amqpwrapper.ts

@@ -4,6 +4,7 @@ import { Util } from "./Util";
 import { Config } from "./Config";
 import { cli } from "winston/lib/winston/config";
 import { Crypt } from "./Crypt";
+import { WebSocketClient } from "./WebSocketClient";
 
 type QueueOnMessage = (msg: string, options: QueueMessageOptions, ack: any, done: any) => void;
 interface IHashTable<T> {
@@ -39,6 +40,7 @@ export class amqpqueue {
     public ok: AssertQueue;
     public QueueOptions: any;
     public consumerTag: string;
+    public cli: WebSocketClient;
 }
 export class amqpexchange {
     public exchange: string;
@@ -48,6 +50,7 @@ export class amqpexchange {
     public callback: QueueOnMessage;
     public ok: amqplib.Replies.AssertExchange;
     public ExchangeOptions: any;
+    public cli: WebSocketClient;
 }
 
 // tslint:disable-next-line: class-name
@@ -144,7 +147,7 @@ export class amqpwrapper {
             if (!Util.IsNullEmpty(this.replyqueue)) {
                 delete this.queues[this.replyqueue];
             }
-            this.replyqueue = await this.AddQueueConsumer("", null, null, (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
+            this.replyqueue = await this.AddQueueConsumer(null, "", null, null, (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
                 if (!Util.IsNullUndefinded(this.activecalls[options.correlationId])) {
                     this.activecalls[options.correlationId].resolve(msg);
                     this.activecalls[options.correlationId] = null;
@@ -203,9 +206,10 @@ export class amqpwrapper {
         if (this.channel != null) await this.channel.cancel(q.consumerTag);
         delete this.queues[q.queue];
     }
-    async AddQueueConsumer(queue: string, QueueOptions: any, jwt: string, callback: QueueOnMessage): Promise<string> {
+    async AddQueueConsumer(cli: WebSocketClient, queue: string, QueueOptions: any, jwt: string, callback: QueueOnMessage): Promise<string> {
         if (this.channel == null || this.conn == null) throw new Error("Cannot Add new Queue Consumer, not connected to rabbitmq");
         var q: amqpqueue = null;
+        q.cli = cli;
         if (Config.amqp_force_queue_prefix && !Util.IsNullEmpty(jwt)) {
             var tuser = Crypt.verityToken(jwt);
             var name = tuser.username.split("@").join("").split(".").join("");
@@ -234,15 +238,16 @@ export class amqpwrapper {
         q.queue = q.ok.queue;
         this._logger.info("[AMQP] Added queue consumer " + q.queue);
         var consumeresult = await this.channel.consume(q.ok.queue, (msg) => {
-            this.OnMessage(this, msg, q.callback);
+            this.OnMessage(q, msg, q.callback);
         }, { noAck: false });
         q.consumerTag = consumeresult.consumerTag;
         this.queues[q.queue] = q;
         return q.queue;
     }
-    async AddExchangeConsumer(exchange: string, algorithm: string, routingkey: string, ExchangeOptions: any, jwt: string, callback: QueueOnMessage): Promise<void> {
+    async AddExchangeConsumer(cli: WebSocketClient, exchange: string, algorithm: string, routingkey: string, ExchangeOptions: any, jwt: string, callback: QueueOnMessage): Promise<void> {
         if (this.channel == null || this.conn == null) throw new Error("Cannot Add new Exchange Consumer, not connected to rabbitmq");
         var q: amqpexchange = null;
+        q.cli = cli;
         if (Config.amqp_force_exchange_prefix && !Util.IsNullEmpty(jwt)) {
             var tuser = Crypt.verityToken(jwt);
             var name = tuser.username.split("@").join("").split(".").join("");
@@ -266,14 +271,27 @@ export class amqpwrapper {
             AssertQueueOptions = Object.create(this.AssertQueueOptions);
             delete AssertQueueOptions.arguments;
         }
-        q.queue = await this.AddQueueConsumer("", AssertQueueOptions, jwt, q.callback);
+        q.queue = await this.AddQueueConsumer(cli, "", AssertQueueOptions, jwt, q.callback);
         this.channel.bindQueue(q.queue, q.exchange, q.routingkey);
         this._logger.info("[AMQP] Added exchange consumer " + q.exchange);
         this.exchanges[exchange] = q;
     }
-    OnMessage(sender: amqpwrapper, msg: amqplib.ConsumeMessage, callback: QueueOnMessage): void {
+    OnMessage(sender: amqpqueue, msg: amqplib.ConsumeMessage, callback: QueueOnMessage): void {
         // sender._logger.info("OnMessage " + msg.content.toString());
         try {
+            var now = new Date();
+            var seconds = (now.getTime() - sender.cli.lastheartbeat.getTime()) / 1000;
+            if (seconds >= 20) {
+                try {
+                    sender.cli._logger.info("amqpwrapper.OnMessage: receive message for inactive client, nack message and try and close");
+                    this.channel.nack(msg);
+                    sender.cli.Close();
+                } catch (error) {
+                    console.error(error);
+                }
+                return;
+            }
+
             var correlationId: string = msg.properties.correlationId;
             var replyTo: string = msg.properties.replyTo;
             var consumerTag: string = msg.fields.consumerTag;

OpenFlow/src/index.ts

@@ -34,7 +34,7 @@ async function initamqp() {
     await testamqp.connect();
 
     // Must also consume messages in the dead letter queue, to catch messages that have timed out
-    await amqp.AddExchangeConsumer(Config.amqp_dlx, "fanout", "", null, null, (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
+    await amqp.AddExchangeConsumer(null, Config.amqp_dlx, "fanout", "", null, null, (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
         // This is the function to run when the dead letter (timed out) message is picked up
         // var data = JSON.parse(msg.content.toString());
         // Change the command and return back to the correct queue (replyTo) to be handled

VERSION

@@ -1 +1 @@
-1.0.11
+1.0.12

docker-compose-toolbox.yml

@@ -45,7 +45,7 @@ services:
       - "traefik.http.routers.web.rule=Host(`toolbox.openrpa.dk`)"
       - "traefik.http.routers.web.entrypoints=web"
       - "traefik.frontend.passHostHeader=true"
-    image: "cloudhack/openflow:1.0.10"
+    image: "cloudhack/openflow:1.0.12"
     container_name: "web"
     environment:
       - update_acl_based_on_groups=true
@@ -82,7 +82,7 @@ services:
       - "traefik.http.routers.nodered.rule=Host(`nodered1.toolbox.openrpa.dk`)"
       - "traefik.http.routers.nodered.entrypoints=web"
       - "traefik.http.services.nodered.loadbalancer.server.port=1880"
-    image: "cloudhack/openflownodered:1.0.10"
+    image: "cloudhack/openflownodered:1.0.12"
     container_name: "nodered"
     environment:
       #      - nodered_id=1

docker-compose-traefik.yml

@@ -45,7 +45,7 @@ services:
       - "traefik.http.routers.web.rule=Host(`localhost.openrpa.dk`)"
       - "traefik.http.routers.web.entrypoints=web"
       - "traefik.frontend.passHostHeader=true"
-    image: "cloudhack/openflow:1.0.10"
+    image: "cloudhack/openflow:1.0.12"
     container_name: "web"
     environment:
       - update_acl_based_on_groups=true
@@ -82,7 +82,7 @@ services:
       - "traefik.http.routers.nodered.rule=Host(`nodered1.localhost.openrpa.dk`)"
       - "traefik.http.routers.nodered.entrypoints=web"
       - "traefik.http.services.nodered.loadbalancer.server.port=1880"
-    image: "cloudhack/openflownodered:1.0.10"
+    image: "cloudhack/openflownodered:1.0.12"
     container_name: "nodered"
     environment:
       #      - nodered_id=1

docker-compose.yml

@@ -18,7 +18,7 @@ services:
       - "5672:5672"
       - "15672:15672"
   web:
-    image: "cloudhack/openflow:1.0.10"
+    image: "cloudhack/openflow:1.0.12"
     environment:
       - update_acl_based_on_groups=true
       - multi_tenant=false
@@ -52,7 +52,7 @@ services:
       - "80:80"
       - "5858:5858"
   nodered:
-    image: "cloudhack/openflownodered:1.0.10"
+    image: "cloudhack/openflownodered:1.0.12"
     environment:
       #      - nodered_id=1
       - nodered_sa=nodered1