Skip to content

Commit c1f21fa

Browse files
skadefroskadefro
committed
try 1.0.22
1 parent 6f806dc commit c1f21fa

8 files changed

Lines changed: 81 additions & 70 deletions

File tree

‎OpenFlow/src/WebSocketClient.ts‎

Lines changed: 23 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ import { DeleteOneMessage } from "./Messages/DeleteOneMessage";
1616
import { Base } from "./base";
1717
import { UpdateManyMessage } from "./Messages/UpdateManyMessage";
1818
import { Util } from "./Util";
19-
import { amqpwrapper, QueueMessageOptions } from "./amqpwrapper";
19+
import { amqpwrapper, QueueMessageOptions, amqpqueue } from "./amqpwrapper";
2020

2121
interface IHashTable<T> {
2222
[key: string]: T;
@@ -47,7 +47,7 @@ export class WebSocketClient {
4747

4848
user: User;
4949
// public consumers: amqp_consumer[] = [];
50-
private queues: IHashTable<string> = {};
50+
private queues: IHashTable<amqpqueue> = {};
5151

5252
constructor(logger: winston.Logger, socketObject: WebSocket) {
5353
this._logger = logger;
@@ -108,7 +108,7 @@ export class WebSocketClient {
108108
public async CloseConsumer(queuename: string): Promise<void> {
109109
if (this.queues[queuename] != null) {
110110
try {
111-
await amqpwrapper.Instance().RemoveQueueConsumer(queuename);
111+
await amqpwrapper.Instance().RemoveQueueConsumer(this.queues[queuename]);
112112
delete this.queues[queuename];
113113
} catch (error) {
114114
this._logger.error("WebSocketclient::CloseConsumer " + error);
@@ -117,24 +117,29 @@ export class WebSocketClient {
117117
}
118118
public async CreateConsumer(queuename: string): Promise<string> {
119119
var autoDelete: boolean = false; // Should we keep the queue around ? for robots and roles
120-
if (Util.IsNullEmpty(queuename)) {
120+
var qname = queuename;
121+
if (Util.IsNullEmpty(qname)) {
121122
if (this.clientagent == "nodered") {
122-
queuename = "nodered." + Math.random().toString(36).substr(2, 9); autoDelete = true;
123+
qname = "nodered." + Math.random().toString(36).substr(2, 9); autoDelete = true;
123124
} else if (this.clientagent == "webapp") {
124-
queuename = "webapp." + Math.random().toString(36).substr(2, 9); autoDelete = true;
125+
qname = "webapp." + Math.random().toString(36).substr(2, 9); autoDelete = true;
125126
} else if (this.clientagent == "web") {
126-
queuename = "web." + Math.random().toString(36).substr(2, 9); autoDelete = true;
127+
qname = "web." + Math.random().toString(36).substr(2, 9); autoDelete = true;
127128
} else {
128-
queuename = "unknown." + Math.random().toString(36).substr(2, 9); autoDelete = true;
129+
qname = "unknown." + Math.random().toString(36).substr(2, 9); autoDelete = true;
129130
}
130131
}
132+
if (this.queues[qname] != null) {
133+
await amqpwrapper.Instance().RemoveQueueConsumer(this.queues[qname]);
134+
delete this.queues[qname];
135+
}
131136
// var AssertQueueOptions: any = new Object(amqpwrapper.Instance().AssertQueueOptions);
132137
var AssertQueueOptions: any = Object.assign({}, (amqpwrapper.Instance().AssertQueueOptions));
133138
AssertQueueOptions.autoDelete = autoDelete;
134-
var queuename = await amqpwrapper.Instance().AddQueueConsumer(queuename, AssertQueueOptions, this.jwt, async (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
139+
var queue = await amqpwrapper.Instance().AddQueueConsumer(qname, AssertQueueOptions, this.jwt, async (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
135140
var _data = msg;
136141
try {
137-
_data = await this.Queue(msg, queuename, options);
142+
_data = await this.Queue(msg, qname, options);
138143
ack();
139144
done(_data);
140145
} catch (error) {
@@ -143,15 +148,19 @@ export class WebSocketClient {
143148
// ack(); // just eat the error
144149
done(_data);
145150
if (error.message != null && error.message != "") {
146-
console.log(queuename + " failed message queue message, nack and re queue message: ", error.message);
151+
console.log(qname + " failed message queue message, nack and re queue message: ", error.message);
147152
} else {
148-
console.log(queuename + " failed message queue message, nack and re queue message: ", error);
153+
console.log(qname + " failed message queue message, nack and re queue message: ", error);
149154
}
150155
}, Config.amqp_requeue_time);
151156
}
152157
});
153-
this.queues[queuename] = queuename;
154-
return queuename;
158+
qname = queue.queue;
159+
if (this.queues[qname] != null) {
160+
await amqpwrapper.Instance().RemoveQueueConsumer(this.queues[qname]);
161+
}
162+
this.queues[qname] = queue;
163+
return queue.queue;
155164
}
156165
sleep(ms) {
157166
return new Promise(resolve => {

‎OpenFlow/src/amqpwrapper.ts‎

Lines changed: 38 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,7 @@ import * as winston from "winston";
22
import * as amqplib from "amqplib";
33
import { Util } from "./Util";
44
import { Config } from "./Config";
5-
import { cli } from "winston/lib/winston/config";
65
import { Crypt } from "./Crypt";
7-
import { WebSocketClient } from "./WebSocketClient";
86

97
type QueueOnMessage = (msg: string, options: QueueMessageOptions, ack: any, done: any) => void;
108
interface IHashTable<T> {
@@ -46,7 +44,7 @@ export class amqpexchange {
4644
public exchange: string;
4745
public algorithm: string;
4846
public routingkey: string;
49-
public queue: string;
47+
public queue: amqpqueue;
5048
public callback: QueueOnMessage;
5149
public ok: amqplib.Replies.AssertExchange;
5250
public ExchangeOptions: any;
@@ -62,9 +60,11 @@ export class amqpwrapper {
6260
public AssertExchangeOptions: any = { durable: false, confirm: true };
6361
public AssertQueueOptions: any = {};
6462
private activecalls: IHashTable<Deferred<string>> = {};
65-
public queues: IHashTable<amqpqueue> = {};
66-
private exchanges: IHashTable<amqpexchange> = {};
67-
private replyqueue: string;
63+
// public queues: IHashTable<amqpqueue> = {};
64+
// private exchanges: IHashTable<amqpexchange> = {};
65+
public queues: amqpqueue[] = [];
66+
private exchanges: amqpexchange[] = [];
67+
private replyqueue: amqpqueue;
6868

6969
private static _instance: amqpwrapper = null;
7070
public static Instance(): amqpwrapper {
@@ -145,7 +145,7 @@ export class amqpwrapper {
145145
}
146146
this.channel = await this.conn.createChannel();
147147
if (!Util.IsNullEmpty(this.replyqueue)) {
148-
delete this.queues[this.replyqueue];
148+
this.queues = this.queues.filter(x => x.consumerTag != this.replyqueue.consumerTag);
149149
}
150150
this.replyqueue = await this.AddQueueConsumer("", null, null, (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
151151
if (!Util.IsNullUndefinded(this.activecalls[options.correlationId])) {
@@ -194,19 +194,13 @@ export class amqpwrapper {
194194
this.timeout = setTimeout(this.connect.bind(this), 1000);
195195
}
196196
}
197-
async RemoveQueueConsumer(queue: string): Promise<string> {
198-
var q: amqpqueue = null;
199-
if (this.queues[queue] != null) {
200-
q = this.queues[queue];
201-
} else {
202-
this._logger.error("[AMQP] Request for removing unknown consumer " + queue);
203-
return;
197+
async RemoveQueueConsumer(queue: amqpqueue): Promise<void> {
198+
if (queue != null) {
199+
this._logger.info("[AMQP] Remove queue consumer " + queue.queue);
200+
if (this.channel != null) await this.channel.cancel(queue.consumerTag);
204201
}
205-
this._logger.info("[AMQP] Remove queue consumer " + queue);
206-
if (this.channel != null) await this.channel.cancel(q.consumerTag);
207-
delete this.queues[q.queue];
208202
}
209-
async AddQueueConsumer(queue: string, QueueOptions: any, jwt: string, callback: QueueOnMessage): Promise<string> {
203+
async AddQueueConsumer(queue: string, QueueOptions: any, jwt: string, callback: QueueOnMessage): Promise<amqpqueue> {
210204
if (this.channel == null || this.conn == null) throw new Error("Cannot Add new Queue Consumer, not connected to rabbitmq");
211205
var q: amqpqueue = null;
212206
if (Config.amqp_force_queue_prefix && !Util.IsNullEmpty(jwt)) {
@@ -217,16 +211,17 @@ export class amqpwrapper {
217211
if (isrole.length == 0 && tuser._id != queue) queue = name + queue;
218212
}
219213

220-
if (this.queues[queue] != null) {
221-
q = this.queues[queue];
222-
try {
223-
if (this.channel != null && !Util.IsNullEmpty(q.consumerTag)) await this.channel.cancel(q.consumerTag);
224-
} catch (error) {
225-
console.error(error);
226-
}
227-
} else {
228-
q = new amqpqueue();
229-
}
214+
// if (this.queues[queue] != null) {
215+
// q = this.queues[queue];
216+
// try {
217+
// if (this.channel != null && !Util.IsNullEmpty(q.consumerTag)) await this.channel.cancel(q.consumerTag);
218+
// } catch (error) {
219+
// console.error(error);
220+
// }
221+
// } else {
222+
// q = new amqpqueue();
223+
// }
224+
q = new amqpqueue();
230225
q.callback = callback;
231226
// q.QueueOptions = new Object((QueueOptions != null ? QueueOptions : this.AssertQueueOptions));
232227
q.QueueOptions = Object.assign({}, (QueueOptions != null ? QueueOptions : this.AssertQueueOptions));
@@ -240,11 +235,12 @@ export class amqpwrapper {
240235
this.OnMessage(q, msg, q.callback);
241236
}, { noAck: false });
242237
q.consumerTag = consumeresult.consumerTag;
243-
this.queues[q.queue] = q;
238+
// this.queues[q.queue] = q;
239+
this.queues.push(q);
244240
this.checkQueue(q.queue);
245-
return q.queue;
241+
return q;
246242
}
247-
async AddExchangeConsumer(exchange: string, algorithm: string, routingkey: string, ExchangeOptions: any, jwt: string, callback: QueueOnMessage): Promise<void> {
243+
async AddExchangeConsumer(exchange: string, algorithm: string, routingkey: string, ExchangeOptions: any, jwt: string, callback: QueueOnMessage): Promise<amqpexchange> {
248244
if (this.channel == null || this.conn == null) throw new Error("Cannot Add new Exchange Consumer, not connected to rabbitmq");
249245
var q: amqpexchange = null;
250246
if (Config.amqp_force_exchange_prefix && !Util.IsNullEmpty(jwt)) {
@@ -253,14 +249,14 @@ export class amqpwrapper {
253249
name = name.toLowerCase();
254250
exchange = name + exchange;
255251
}
256-
257-
if (this.exchanges[exchange] != null) {
258-
q = this.exchanges[exchange];
259-
} else {
260-
q = new amqpexchange();
261-
}
252+
q = new amqpexchange();
253+
// if (this.exchanges[exchange] != null) {
254+
// q = this.exchanges[exchange];
255+
// } else {
256+
// q = new amqpexchange();
257+
// }
262258
if (!Util.IsNullEmpty(q.queue)) {
263-
delete this.queues[q.queue];
259+
this.RemoveQueueConsumer(q.queue);
264260
}
265261
// q.ExchangeOptions = new Object((ExchangeOptions != null ? ExchangeOptions : this.AssertExchangeOptions));
266262
q.ExchangeOptions = Object.assign({}, (ExchangeOptions != null ? ExchangeOptions : this.AssertExchangeOptions));
@@ -274,7 +270,9 @@ export class amqpwrapper {
274270
q.queue = await this.AddQueueConsumer("", AssertQueueOptions, jwt, q.callback);
275271
this.channel.bindQueue(q.queue, q.exchange, q.routingkey);
276272
this._logger.info("[AMQP] Added exchange consumer " + q.exchange);
277-
this.exchanges[exchange] = q;
273+
// this.exchanges[exchange] = q;
274+
this.exchanges.push(q);
275+
return q;
278276
}
279277
OnMessage(sender: amqpqueue, msg: amqplib.ConsumeMessage, callback: QueueOnMessage): void {
280278
// sender._logger.info("OnMessage " + msg.content.toString());
@@ -335,7 +333,7 @@ export class amqpwrapper {
335333
async sendWithReply(exchange: string, queue: string, data: any, expiration: number, correlationId: string): Promise<string> {
336334
if (Util.IsNullEmpty(correlationId)) correlationId = this.generateUuid();
337335
this.activecalls[correlationId] = new Deferred();
338-
await this.sendWithReplyTo(exchange, queue, this.replyqueue, data, expiration, correlationId);
336+
await this.sendWithReplyTo(exchange, queue, this.replyqueue.queue, data, expiration, correlationId);
339337
return this.activecalls[correlationId].promise;
340338
}
341339
async sendWithReplyTo(exchange: string, queue: string, replyTo: string, data: any, expiration: number, correlationId: string): Promise<void> {

‎OpenFlow/src/public/CommonControllers.ts‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -223,12 +223,16 @@ module openflow {
223223
var msg: Message = new Message(); msg.command = "deleteone"; msg.data = JSON.stringify(q);
224224
q = await this.WebSocketClient.Send<DeleteOneMessage>(msg);
225225
}
226+
private queuename: string = "";
226227
async RegisterQueue(queuename: string = undefined): Promise<string> {
227-
var q: RegisterQueueMessage = new RegisterQueueMessage();
228-
q.queuename = queuename;
229-
var msg: Message = new Message(); msg.command = "registerqueue"; msg.data = JSON.stringify(q);
230-
var result: RegisterQueueMessage = await this.WebSocketClient.Send(msg);
231-
return result.queuename;
228+
if (this.queuename == "") {
229+
var q: RegisterQueueMessage = new RegisterQueueMessage();
230+
q.queuename = queuename;
231+
var msg: Message = new Message(); msg.command = "registerqueue"; msg.data = JSON.stringify(q);
232+
var result: RegisterQueueMessage = await this.WebSocketClient.Send(msg);
233+
this.queuename = result.queuename;
234+
}
235+
return this.queuename;
232236
}
233237
async _QueueMessage(queuename: string, replyto: string, data: any, expiration: number): Promise<QueueMessage> {
234238
return new Promise<QueueMessage>(async (resolve, reject) => {

‎OpenFlowNodeRED/src/nodered/nodes/rpa_nodes.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,7 @@ export class rpa_workflow_node {
119119
this.localqueue = await NoderedUtil.RegisterQueue(WebSocketClient.instance, this.localqueue, (msg: QueueMessage, ack: any) => {
120120
this.OnMessage(msg, ack);
121121
});
122-
this.node.status({ fill: "green", shape: "dot", text: "Connected" });
122+
this.node.status({ fill: "green", shape: "dot", text: "Connected " + this.localqueue });
123123

124124
} catch (error) {
125125
NoderedUtil.HandleError(this, error);
@@ -159,7 +159,7 @@ export class rpa_workflow_node {
159159
// if (!NoderedUtil.IsNullEmpty(data.jwt)) { result.jwt = data.jwt; }
160160
if (data.user != null) result.user = data.user;
161161
if (result.payload == null || result.payload == undefined) { result.payload = {}; }
162-
this.node.status({ fill: "green", shape: "dot", text: command });
162+
this.node.status({ fill: "green", shape: "dot", text: command + " " + this.localqueue });
163163
this.node.send(result);
164164
}
165165
else if (command == "invokefailed" || command == "invokeaborted" || command == "error" || command == "timeout") {
@@ -174,7 +174,7 @@ export class rpa_workflow_node {
174174
// if (!NoderedUtil.IsNullEmpty(data.jwt)) { result.jwt = data.jwt; }
175175
if (data.user != null) result.user = data.user;
176176
if (result.payload == null || result.payload == undefined) { result.payload = {}; }
177-
this.node.status({ fill: "red", shape: "dot", text: command });
177+
this.node.status({ fill: "red", shape: "dot", text: command + " " + this.localqueue });
178178
this.node.send([null, null, result]);
179179
}
180180
else {
@@ -225,7 +225,7 @@ export class rpa_workflow_node {
225225
}
226226
// this.con.SendMessage(JSON.stringify(rpacommand), targetid, correlationId, true);
227227
await NoderedUtil.QueueMessage(WebSocketClient.instance, targetid, this.localqueue, rpacommand, correlationId, expiration);
228-
this.node.status({ fill: "blue", shape: "dot", text: "Robot running..." });
228+
this.node.status({ fill: "blue", shape: "dot", text: "Robot running " + this.localqueue });
229229
} catch (error) {
230230
// NoderedUtil.HandleError(this, error);
231231
try {

‎VERSION‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
1.0.21
1+
1.0.22

‎docker-compose-toolbox.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ services:
4545
- "traefik.http.routers.web.rule=Host(`toolbox.openrpa.dk`)"
4646
- "traefik.http.routers.web.entrypoints=web"
4747
- "traefik.frontend.passHostHeader=true"
48-
image: "cloudhack/openflow:1.0.15"
48+
image: "cloudhack/openflow:1.0.22"
4949
container_name: "web"
5050
environment:
5151
- update_acl_based_on_groups=true
@@ -82,7 +82,7 @@ services:
8282
- "traefik.http.routers.nodered.rule=Host(`nodered1.toolbox.openrpa.dk`)"
8383
- "traefik.http.routers.nodered.entrypoints=web"
8484
- "traefik.http.services.nodered.loadbalancer.server.port=1880"
85-
image: "cloudhack/openflownodered:1.0.15"
85+
image: "cloudhack/openflownodered:1.0.22"
8686
container_name: "nodered"
8787
environment:
8888
# - nodered_id=1

‎docker-compose-traefik.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ services:
4545
- "traefik.http.routers.web.rule=Host(`localhost.openrpa.dk`)"
4646
- "traefik.http.routers.web.entrypoints=web"
4747
- "traefik.frontend.passHostHeader=true"
48-
image: "cloudhack/openflow:1.0.15"
48+
image: "cloudhack/openflow:1.0.22"
4949
container_name: "web"
5050
environment:
5151
- update_acl_based_on_groups=true
@@ -82,7 +82,7 @@ services:
8282
- "traefik.http.routers.nodered.rule=Host(`nodered1.localhost.openrpa.dk`)"
8383
- "traefik.http.routers.nodered.entrypoints=web"
8484
- "traefik.http.services.nodered.loadbalancer.server.port=1880"
85-
image: "cloudhack/openflownodered:1.0.15"
85+
image: "cloudhack/openflownodered:1.0.22"
8686
container_name: "nodered"
8787
environment:
8888
# - nodered_id=1

‎docker-compose.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ services:
1818
- "5672:5672"
1919
- "15672:15672"
2020
web:
21-
image: "cloudhack/openflow:1.0.15"
21+
image: "cloudhack/openflow:1.0.22"
2222
environment:
2323
- update_acl_based_on_groups=true
2424
- multi_tenant=false
@@ -52,7 +52,7 @@ services:
5252
- "80:80"
5353
- "5858:5858"
5454
nodered:
55-
image: "cloudhack/openflownodered:1.0.15"
55+
image: "cloudhack/openflownodered:1.0.22"
5656
environment:
5757
# - nodered_id=1
5858
- nodered_sa=nodered1

0 commit comments

Comments
 (0)