Add amqp_enabled_exchange option

Author: skadefro

OpenFlow/src/Config.ts

@@ -82,6 +82,7 @@ export class Config {
         Config.amqp_force_queue_prefix = Config.parseBoolean(Config.getEnv("amqp_force_queue_prefix", "false"));
         Config.amqp_force_exchange_prefix = Config.parseBoolean(Config.getEnv("amqp_force_exchange_prefix", "false"));
         Config.amqp_force_sender_has_read = Config.parseBoolean(Config.getEnv("amqp_force_sender_has_read", "true"));
+        Config.amqp_enabled_exchange = Config.parseBoolean(Config.getEnv("amqp_enabled_exchange", "false"));
         Config.amqp_url = Config.getEnv("amqp_url", "amqp://localhost"); // used to register queues and by personal nodered
         Config.amqp_check_for_consumer = Config.parseBoolean(Config.getEnv("amqp_check_for_consumer", "true"));
         Config.amqp_check_for_consumer_count = Config.parseBoolean(Config.getEnv("amqp_check_for_consumer_count", "false"));
@@ -211,6 +212,7 @@ export class Config {
     public static amqp_force_queue_prefix: boolean = Config.parseBoolean(Config.getEnv("amqp_force_queue_prefix", "false"));
     public static amqp_force_exchange_prefix: boolean = Config.parseBoolean(Config.getEnv("amqp_force_exchange_prefix", "false"));
     public static amqp_force_sender_has_read: boolean = Config.parseBoolean(Config.getEnv("amqp_force_sender_has_read", "true"));
+    public static amqp_enabled_exchange: boolean = Config.parseBoolean(Config.getEnv("amqp_enabled_exchange", "false"));
     public static amqp_url: string = Config.getEnv("amqp_url", "amqp://localhost"); // used to register queues and by personal nodered
     public static amqp_check_for_consumer: boolean = Config.parseBoolean(Config.getEnv("amqp_check_for_consumer", "true"));
     public static amqp_check_for_consumer_count: boolean = Config.parseBoolean(Config.getEnv("amqp_check_for_consumer_count", "false"));

OpenFlow/src/LoginProvider.ts

@@ -421,7 +421,8 @@ export class LoginProvider {
                     getting_started_url: Config.getting_started_url,
                     validate_user_form: Config.validate_user_form,
                     supports_watch: Config.supports_watch,
-                    nodered_images: Config.nodered_images
+                    nodered_images: Config.nodered_images,
+                    amqp_enabled_exchange: Config.amqp_enabled_exchange
                 }
                 res.end(JSON.stringify(res2));
             } catch (error) {

OpenFlow/src/Messages/Message.ts

@@ -360,6 +360,9 @@ export class Message {
                     msg.data.jwt = msg.jwt;
                 }
             }
+            if (!NoderedUtil.IsNullEmpty(msg.exchange) && !Config.amqp_enabled_exchange) {
+                throw new Error("AMQP exchange is not enabled on this OpenFlow");
+            }
             const expiration: number = (typeof msg.expiration == 'number' ? msg.expiration : Config.amqp_default_expiration);
             if (typeof msg.data === 'string' || msg.data instanceof String) {
                 try {
@@ -1450,6 +1453,7 @@ export class Message {
                     "otel_metric_url=" + Config.otel_metric_url,
                     "otel_trace_interval=" + Config.otel_trace_interval.toString(),
                     "otel_metric_interval=" + Config.otel_metric_interval.toString(),
+                    "amqp_enabled_exchange=" + Config.amqp_enabled_exchange.toString()
                 ]
                 // const image = await docker.pull(nodered_image, { serveraddress: "https://index.docker.io/v1" });
                 await this._pullImage(docker, nodered_image);
@@ -1702,7 +1706,7 @@ export class Message {
                                             { name: "otel_metric_url", value: Config.otel_metric_url },
                                             { name: "otel_trace_interval", value: Config.otel_trace_interval.toString() },
                                             { name: "otel_metric_interval", value: Config.otel_metric_interval.toString() },
-
+                                            { name: "amqp_enabled_exchange", value: Config.amqp_enabled_exchange.toString() },
                                         ],
                                         livenessProbe: livenessProbe,
                                     }

OpenFlow/src/amqpwrapper.ts

@@ -265,6 +265,7 @@ export class amqpwrapper extends events.EventEmitter {
                     let name = tuser.username.split("@").join("").split(".").join("");
                     name = name.toLowerCase();
                     queue = name + queue;
+                    if (queue.length == 24) { queue += "1"; }
                 }
             } else if (queue.length == 24) {
                 if (NoderedUtil.IsNullEmpty(jwt)) {
@@ -327,6 +328,7 @@ export class amqpwrapper extends events.EventEmitter {
                 let name = tuser.username.split("@").join("").split(".").join("");
                 name = name.toLowerCase();
                 exchange = name + exchange;
+                if (exchange.length == 24) { exchange += "1"; }
             }
             const q: amqpexchange = new amqpexchange();
             if (!NoderedUtil.IsNullEmpty(q.queue)) {

OpenFlowNodeRED/src/Config.ts

@@ -49,7 +49,8 @@ export class Config {
         Config.amqp_reply_expiration = parseInt(Config.getEnv("amqp_reply_expiration", (60 * 1000).toString())); // 1 min
         Config.amqp_workflow_out_expiration = parseInt(Config.getEnv("amqp_workflow_out_expiration", (60 * 1000).toString())); // 1 min
         Config.amqp_reply_expiration = parseInt(Config.getEnv("amqp_reply_expiration", "10000")); // 10 seconds
-        Config.amqp_workflow_out_expiration = parseInt(Config.getEnv("amqp_workflow_out_expiration", "10000")); // 10 seconds
+        Config.amqp_enabled_exchange = Config.parseBoolean(Config.getEnv("amqp_enabled_exchange", "false"));
+
 
         Config.api_credential_cache_seconds = parseInt(Config.getEnv("api_credential_cache_seconds", "300"));
         Config.api_allow_anonymous = Config.parseBoolean(Config.getEnv("api_allow_anonymous", "false"));
@@ -115,6 +116,7 @@ export class Config {
     // public static amqp_workflow_out_expiration: number = parseInt(Config.getEnv("amqp_workflow_out_expiration", (60 * 1000).toString())); // 1 min
     public static amqp_reply_expiration: number = parseInt(Config.getEnv("amqp_reply_expiration", "10000")); // 10 seconds
     public static amqp_workflow_out_expiration: number = parseInt(Config.getEnv("amqp_workflow_out_expiration", "10000")); // 10 seconds
+    public static amqp_enabled_exchange: boolean = Config.parseBoolean(Config.getEnv("amqp_enabled_exchange", "false"));
 
     public static api_credential_cache_seconds: number = parseInt(Config.getEnv("api_credential_cache_seconds", "300"));
     public static api_allow_anonymous: boolean = Config.parseBoolean(Config.getEnv("api_allow_anonymous", "false"));

OpenFlowNodeRED/src/nodered/nodes/amqp.ts

@@ -1,4 +1,5 @@
 import { Red } from "node-red";
+import { Config } from "../../Config";
 import * as amqp from "./amqp_nodes";
 
 
@@ -14,5 +15,7 @@ export = function (RED: Red) {
     RED.nodes.registerType("amqp consumer", amqp.amqp_consumer_node);
     RED.nodes.registerType("amqp publisher", amqp.amqp_publisher_node);
     RED.nodes.registerType("amqp acknowledgment", amqp.amqp_acknowledgment_node);
-    RED.nodes.registerType("amqp exchange", amqp.amqp_exchange_node);
+    if (Config.amqp_enabled_exchange) {
+        RED.nodes.registerType("amqp exchange", amqp.amqp_exchange_node);
+    }
 }