improve queue info/control

skadefro · skadefro · commit c910c8b54886 · 2020-07-26T13:21:45.000+02:00
diff --git a/OpenFlow/src/Messages/Message.ts b/OpenFlow/src/Messages/Message.ts
@@ -182,6 +182,12 @@ export class Message {
                 case "dumprabbitmq":
                     this.DumpRabbitmq(cli);
                     break;
+                case "getrabbitmqqueue":
+                    this.GetRabbitmqQueue(cli);
+                    break;
+                case "deleterabbitmqqueue":
+                    this.DeleterabbitmqQueue(cli);
+                    break;
                 default:
                     this.UnknownCommand(cli);
                     break;
@@ -2229,6 +2235,13 @@ export class Message {
                         consumers = queue.consumer_details.length;
                     }
                 }
+                // if (consumers > 0 && queue.consumer_details == null) {
+                //     var tempconfig = await amqpwrapper.getqueue(Config.amqp_url, '/', queue.name);
+                //     // var tempconfig = await amqpwrapper.getqueue(queue.name);
+                //     if (tempconfig.consumer_details != null) {
+                //         item.consumer_details = tempconfig.consumer_details
+                //     }
+                // }
                 item.queuename = queue.name;
                 item.consumers = consumers;
                 item.name = queue.name + "(" + consumers + ")";
@@ -2265,6 +2278,45 @@ export class Message {
         }
         this.Send(cli);
     }
+    async GetRabbitmqQueue(cli: WebSocketServerClient) {
+        this.Reply();
+        try {
+            let msg: any = JSON.parse(this.data);
+            const kickstartapi = amqpwrapper.getvhosts(Config.amqp_url);
+            try {
+                msg.data = await amqpwrapper.getqueue(Config.amqp_url, '/', msg.name);
+                this.data = JSON.stringify(msg);
+            } catch (error) {
+                cli._logger.error(error);
+            }
+        } catch (error) {
+            this.command = "error";
+            this.data = JSON.stringify(error);
+            cli._logger.error(error);
+
+        }
+        this.Send(cli);
+    }
+    async DeleterabbitmqQueue(cli: WebSocketServerClient) {
+        this.Reply();
+        try {
+            let msg: any = JSON.parse(this.data);
+            const kickstartapi = amqpwrapper.getvhosts(Config.amqp_url);
+            try {
+                msg.data = await amqpwrapper.deletequeue(Config.amqp_url, '/', msg.name);
+                this.data = JSON.stringify(msg);
+            } catch (error) {
+                cli._logger.error(error);
+            }
+        } catch (error) {
+            this.command = "error";
+            this.data = JSON.stringify(error);
+            cli._logger.error(error);
+
+        }
+        this.Send(cli);
+    }
+
 }
 
 export class JSONfn {
diff --git a/OpenFlow/src/amqpwrapper.ts b/OpenFlow/src/amqpwrapper.ts
@@ -402,8 +402,8 @@ export class amqpwrapper {
         var result: boolean = false;
         try {
             result = await retry(async bail => {
-                // var queue = await amqpwrapper.getqueue(Config.amqp_url, '/', queuename);
-                var queue = await amqpwrapper.getqueue(queuename);
+                var queue = await amqpwrapper.getqueue(Config.amqp_url, '/', queuename);
+                // var queue = await amqpwrapper.getqueue(queuename);
                 let hasConsumers: boolean = false;
                 if (queue.consumers > 0) {
                     hasConsumers = true;
@@ -466,29 +466,45 @@ export class amqpwrapper {
         var payload = JSON.parse(response.body);
         return payload;
     }
-    static async getqueue(queuename) {
-        const queues = await amqpwrapper.getqueues(Config.amqp_url);
-        for (let i = 0; i < queues.length; i++) {
-            let queue = queues[i];
-            if (queue.name == queuename) {
-                return queue;
-            }
-        }
-        // var q = this.parseurl(amqp_url);
-        // var options = {
-        //     headers: {
-        //         'Content-type': 'application/x-www-form-urlencoded'
-        //     },
-        //     username: (q as any).username,
-        //     password: (q as any).password,
-        //     timeout: 500, retry: 1
-        // };
-        // var _url = 'http://' + q.host + ':' + q.port + '/api/queues/' + encodeURIComponent(vhost) + '/' + queuename;
-        // var response = await got.get(_url, options);
-        // var payload = JSON.parse(response.body);
-        // return payload;
+    static async getqueue(amqp_url: string, vhost: string, queuename) {
+        // const queues = await amqpwrapper.getqueues(Config.amqp_url);
+        // for (let i = 0; i < queues.length; i++) {
+        //     let queue = queues[i];
+        //     if (queue.name == queuename) {
+        //         return queue;
+        //     }
+        // }
+        var q = this.parseurl(amqp_url);
+        var options = {
+            headers: {
+                'Content-type': 'application/x-www-form-urlencoded'
+            },
+            username: (q as any).username,
+            password: (q as any).password,
+            timeout: 500, retry: 1
+        };
+        var _url = 'http://' + q.host + ':' + q.port + '/api/queues/' + encodeURIComponent(vhost) + '/' + encodeURIComponent(queuename);
+        var response = await got.get(_url, options);
+        var payload = JSON.parse(response.body);
+        return payload;
+    }
+    static async deletequeue(amqp_url: string, vhost: string, queuename) {
+        var q = this.parseurl(amqp_url);
+        var options = {
+            headers: {
+                'Content-type': 'application/x-www-form-urlencoded'
+            },
+            username: (q as any).username,
+            password: (q as any).password,
+            timeout: 500, retry: 1
+        };
+        var _url = 'http://' + q.host + ':' + q.port + '/api/queues/' + encodeURIComponent(vhost) + '/' + encodeURIComponent(queuename);
+        var response = await got.delete(_url, options);
+        var payload = JSON.parse(response.body);
+        return payload;
     }
 
+
     // async checkQueue(queue: string): Promise<boolean> {
     //     if (Config.amqp_check_for_consumer) {
     //         var q: amqpqueue = this.queues[queue];
diff --git a/OpenFlow/src/public/Controllers.ts b/OpenFlow/src/public/Controllers.ts
@@ -3453,6 +3453,107 @@ export class QueuesCtrl extends entitiesCtrl<Base> {
         this.loadData();
     }
 }
+export class QueueCtrl extends entityCtrl<Base> {
+    public newid: string;
+    public memberof: Role[];
+    public data: any;
+    constructor(
+        public $scope: ng.IScope,
+        public $location: ng.ILocationService,
+        public $routeParams: ng.route.IRouteParamsService,
+        public $interval: ng.IIntervalService,
+        public WebSocketClientService: WebSocketClientService,
+        public api: api
+    ) {
+        super($scope, $location, $routeParams, $interval, WebSocketClientService, api);
+        console.debug("QueueCtrl");
+        this.collection = "configclients";
+        this.postloadData = this.processdata;
+        this.memberof = [];
+        WebSocketClientService.onSignedin((user: TokenUser) => {
+            if (this.id !== null && this.id !== undefined) {
+                this.loadData();
+            } else {
+                this.model = new Base();
+                this.model._type = "queue";
+                this.model.name = "";
+                this.processdata();
+            }
+
+        });
+    }
+    async processdata() {
+        try {
+            if (this.model == null) {
+                this.errormessage = "Not found!";
+                this.loading = false;
+                if (!this.$scope.$$phase) { this.$scope.$apply(); }
+                return;
+            }
+            this.loading = true;
+            let m: Message = new Message();
+            m.command = "getrabbitmqqueue"; m.data = "{\"name\": \"" + (this.model as any).queuename + "\"}";
+            var q = await WebSocketClient.instance.Send<any>(m);
+            if ((q as any).command == "error") throw new Error(q.data);
+            this.data = q.data;
+            if (this.data == null) {
+                this.errormessage = "Queue not found!";
+                this.loading = false;
+                if (!this.$scope.$$phase) { this.$scope.$apply(); }
+                return;
+            }
+            if (this.data.consumer_details == null || this.data.consumer_details.length == 0) {
+                this.errormessage = "Queue has no consumers!";
+                this.loading = false;
+                if (!this.$scope.$$phase) { this.$scope.$apply(); }
+                return;
+            }
+            this.collection = "configclients";
+            this.basequery = { _type: "socketclient" };
+            var clients = await NoderedUtil.Query("configclients", { _type: "socketclient" }, null, null, 2000, 0, null, null);
+            for (var i = 0; i < this.data.consumer_details.length; i++) {
+                console.log("find " + this.data.consumer_details[i].consumer_tag);
+
+                for (var y = 0; y < clients.length; y++) {
+                    const _client = clients[y];
+                    if (_client.queues != null) {
+                        const keys = Object.keys(_client.queues);
+                        for (var z = 0; z < keys.length; z++) {
+                            var q = _client.queues[keys[z]];
+                            console.log(_client.name + " " + q.consumerTag);
+                            if (q.consumerTag == this.data.consumer_details[i].consumer_tag) {
+                                this.data.consumer_details[i].clientname = _client.name;
+                            }
+                        }
+                    }
+
+                }
+
+            }
+        } catch (error) {
+            console.error(error);
+        }
+        this.loading = false;
+        if (!this.$scope.$$phase) { this.$scope.$apply(); }
+    }
+    async DeleteQueue(model) {
+        try {
+            this.loading = true;
+            let m: Message = new Message();
+            m.command = "deleterabbitmqqueue"; m.data = "{\"name\": \"" + (this.model as any).queuename + "\"}";
+            var q = await WebSocketClient.instance.Send<any>(m);
+            if ((q as any).command == "error") throw new Error(q.data);
+            this.data = q.data;
+            this.$location.path("/Queues");
+        } catch (error) {
+            this.errormessage = error;
+            console.error(error);
+        }
+        this.loading = false;
+        if (!this.$scope.$$phase) { this.$scope.$apply(); }
+    }
+
+}
 export class SocketsCtrl extends entitiesCtrl<Base> {
     public message: string = "";
     public charts: chartset[] = [];
diff --git a/OpenFlow/src/public/Queue.html b/OpenFlow/src/public/Queue.html
@@ -0,0 +1,29 @@
+<a ng-disabled="ctrl.loading==true" type="button" class="btn btn-success" ng-href="#/Queues">Queues</a>
+<a ng-disabled="ctrl.loading==true" type="button" class="btn btn-success" ng-href="#/Sockets">Sockets</a>
+<div class="row">
+  <h1 translate lib="web">{{ctrl.model.name}}</h1>
+</div>
+<div ng-show="ctrl.errormessage != ''"" class=" alert alert-danger" role="alert">{{ctrl.errormessage}}</div>
+
+<table id=" table1" class="table table-striped table-hover table-sm" when-scrolled="ctrl.more()" style="width: 100%;">
+  <thead class="thead-dark">
+    <tr>
+      <th scope="col"><b translate lib="web">ack</b></th>
+      <th scope="col"><b translate lib="web">active</b></th>
+      <th scope="col"><b translate lib="web">consumer_tag</b></th>
+      <th scope="col"><b translate lib="web">clientname</b></th>
+      <th scope="col"><b translate lib="web"></b></th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr ng-repeat="model in ctrl.data.consumer_details">
+      <td>{{model.ack_required}}</td>
+      <td>{{model.active}}</td>
+      <td>{{model.consumer_tag}}</td>
+      <td>{{model.clientname}}</td>
+      <td class="btn-cell">
+        <a href ng-click="ctrl.DeleteQueue(model)" ng-disabled="ctrl.loading==true"><i class="az-trash"></i></a>
+      </td>
+    </tr>
+  </tbody>
+</table>
diff --git a/OpenFlow/src/public/Queues.html b/OpenFlow/src/public/Queues.html
@@ -16,7 +16,7 @@
   </thead>
   <tbody>
     <tr ng-repeat="model in ctrl.models">
-      <td>{{model.name}}</td>
+      <td><a ng-href="#/Queue/{{model._id}}">{{model.name}}</a></td>
       <td><span translate lib="web">{{model.type || model._type}}</span></td>
       <td>{{model.consumers}}</td>
       <td>{{model.queuename}}</td>
diff --git a/OpenFlow/src/public/app.ts b/OpenFlow/src/public/app.ts
@@ -1,7 +1,7 @@
 import { WebSocketClientService } from "./WebSocketClientService";
 import angular = require("angular");
 import { timesince, translate, textarea, fileread, userdata, api } from "./CommonControllers";
-import { MenuCtrl, ProvidersCtrl, MainCtrl, LoginCtrl, ProviderCtrl, UsersCtrl, UserCtrl, RolesCtrl, RoleCtrl, RPAWorkflowsCtrl, RPAWorkflowCtrl, WorkflowsCtrl, ReportsCtrl, jslogCtrl, EditFormCtrl, FormsCtrl, FormCtrl, FilesCtrl, EntitiesCtrl, EntityCtrl, HistoryCtrl, SocketCtrl, NoderedCtrl, hdrobotsCtrl, RobotsCtrl, AuditlogsCtrl, SignupCtrl, PaymentCtrl, QueuesCtrl, SocketsCtrl } from "./Controllers";
+import { MenuCtrl, ProvidersCtrl, MainCtrl, LoginCtrl, ProviderCtrl, UsersCtrl, UserCtrl, RolesCtrl, RoleCtrl, RPAWorkflowsCtrl, RPAWorkflowCtrl, WorkflowsCtrl, ReportsCtrl, jslogCtrl, EditFormCtrl, FormsCtrl, FormCtrl, FilesCtrl, EntitiesCtrl, EntityCtrl, HistoryCtrl, SocketCtrl, NoderedCtrl, hdrobotsCtrl, RobotsCtrl, AuditlogsCtrl, SignupCtrl, PaymentCtrl, QueuesCtrl, SocketsCtrl, QueueCtrl } from "./Controllers";
 
 /**
  * @type {angular.Module}
@@ -86,6 +86,8 @@ module openflow {
                 .when('/Payment', { templateUrl: 'Payment.html', controller: PaymentCtrl, controllerAs: 'ctrl' })
                 .when('/Payment/:userid', { templateUrl: 'Payment.html', controller: PaymentCtrl, controllerAs: 'ctrl' })
 
+                .when('/Queue', { templateUrl: 'Queue.html', controller: QueueCtrl, controllerAs: 'ctrl' })
+                .when('/Queue/:id', { templateUrl: 'Queue.html', controller: QueueCtrl, controllerAs: 'ctrl' })
                 .when('/Queues', { templateUrl: 'Queues.html', controller: QueuesCtrl, controllerAs: 'ctrl' })
                 .when('/Sockets', { templateUrl: 'Sockets.html', controller: SocketsCtrl, controllerAs: 'ctrl' })
 
diff --git a/OpenFlowNodeRED/package.json b/OpenFlowNodeRED/package.json
@@ -1,6 +1,6 @@
 {
   "name": "openflow-nodered",
-  "version": "1.0.71",
+  "version": "1.0.72",
   "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": {
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-1.0.71
+1.0.72
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "openiap",
-  "version": "1.0.71",
+  "version": "1.0.72",
   "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": {

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "openflow-nodered",`
`3`		`- "version": "1.0.71",`
	`3`	`+ "version": "1.0.72",`
`4`	`4`	`"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)",`
`5`	`5`	`"main": "index.js",`
`6`	`6`	`"scripts": {`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "openiap",`
`3`		`- "version": "1.0.71",`
	`3`	`+ "version": "1.0.72",`
`4`	`4`	`"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)",`
`5`	`5`	`"main": "index.js",`
`6`	`6`	`"scripts": {`