Dynamic MQTT connections (#3189)

* add mqtt-control
- adds auto-connect option to broker
- add new node mqtt-control
- adds i18n messages
- adds documentation

* documentation tweaks

* built in documentation improvements

* fix tip layout causing oversized editor

* remove unused requires

* add missing `unsubscribe` dropdown option
- oddly forgotten - now added

* ensure clientid is updated dynamically

* [rewrite] move mqtt-control login into mqtt-in

* Remove dynamic label

* remove redundant mqtt-control code left overs

* Callback for brokerConn.connect (improve done())
- done is now called on connect callback

* fix race condition if connect/disconnect too fast
- node.connected and node.client.connected getting out of sync

* fix connection fail when switching protocol 3 ~ 5
- ensure protocolId is correct for protocolVersion

* change msg.subscribe prop to `msg.topic`

* unsub all topics if msg.topic is `true`

* delete temprary debugger statements

* Final rework of dynamic mqtt connections

Co-authored-by: Steve-Mcl <sdmclaughlin@gmail.com>

packages/node_modules/@node-red/nodes/core/network/10-mqtt.html

@@ -54,6 +54,18 @@
         width: 15px;
         height: 15px;
     }
+    .form-row-mqtt5 {
+        display: none;
+    }
+    .form-row-mqtt5.form-row-mqtt5-active:not(.form-row-mqtt-static-disabled) {
+        display: block
+    }
+    .form-row-mqtt-static-disabled {
+        display: none;
+        /* opacity: 0.3;
+        pointer-events: none; */
+    }
+
 </style>
 
 <script type="text/html" data-template-name="mqtt in">
@@ -62,10 +74,18 @@
         <input type="text" id="node-input-broker">
     </div>
     <div class="form-row">
+        <label for="node-input-topicType" data-i18n="mqtt.label.action"></label>
+        <select id="node-input-topicType" style="width: 70%">
+            <option value="topic" data-i18n="mqtt.label.staticTopic"></option>
+            <option value="dynamic" data-i18n="mqtt.label.dynamicTopic"></option>
+        </select>
+        <input type="hidden" id="node-input-inputs">
+    </div>
+    <div class="form-row form-row-mqtt-static">
         <label for="node-input-topic"><i class="fa fa-tasks"></i> <span data-i18n="common.label.topic"></span></label>
         <input type="text" id="node-input-topic" data-i18n="[placeholder]common.label.topic">
     </div>
-    <div class="form-row">
+    <div class="form-row form-row-mqtt-static">
         <label for="node-input-qos"><i class="fa fa-empire"></i> <span data-i18n="mqtt.label.qos"></span></label>
         <select id="node-input-qos" style="width:125px !important">
             <option value="0">0</option>
@@ -73,17 +93,7 @@
             <option value="2">2</option>
         </select>
     </div>
-    <div class="form-row">
+    <div class="form-row mqtt-flags-row form-row-mqtt5 form-row-mqtt-static">
-        <label for="node-input-datatype"><i class="fa fa-sign-out"></i> <span data-i18n="mqtt.label.output"></span></label>
-        <select id="node-input-datatype" style="width:70%;">
-            <option value="auto" data-i18n="mqtt.output.auto"></option>
-            <option value="buffer" data-i18n="mqtt.output.buffer"></option>
-            <option value="utf8" data-i18n="mqtt.output.string"></option>
-            <option value="json" data-i18n="mqtt.output.json"></option>
-            <option value="base64" data-i18n="mqtt.output.base64"></option>
-        </select>
-    </div>
-    <div class="form-row mqtt-flags-row mqtt5">
         <label for="node-input-nl" ><i class="fa fa-flag"></i> <span data-i18n="mqtt.label.flags">Flags</span></label>
         <div class="mqtt-flags">
             <div class="mqtt-flag">
@@ -100,7 +110,7 @@
             </div>
         </div>
     </div>
-    <div class="form-row mqtt5">
+    <div class="form-row form-row-mqtt5 form-row-mqtt-static">
         <label for="node-input-rh" style="width:100%"><i class="fa fa-tag"></i> <span data-i18n="mqtt.label.rh"></span></label>
         <select id="node-input-rh" style="margin-left: 104px; width: 70%">
             <option value="0" data-i18n="mqtt.label.rh0"></option>
@@ -108,6 +118,16 @@
             <option value="2" data-i18n="mqtt.label.rh2"></option>
         </select>
     </div>
+    <div class="form-row">
+        <label for="node-input-datatype"><i class="fa fa-sign-out"></i> <span data-i18n="mqtt.label.output"></span></label>
+        <select id="node-input-datatype" style="width:70%;">
+            <option value="auto" data-i18n="mqtt.output.auto"></option>
+            <option value="buffer" data-i18n="mqtt.output.buffer"></option>
+            <option value="utf8" data-i18n="mqtt.output.string"></option>
+            <option value="json" data-i18n="mqtt.output.json"></option>
+            <option value="base64" data-i18n="mqtt.output.base64"></option>
+        </select>
+    </div>
     <div class="form-row">
         <label for="node-input-name"><i class="fa fa-tag"></i> <span data-i18n="common.label.name"></span></label>
         <input type="text" id="node-input-name" data-i18n="[placeholder]common.label.name">
@@ -185,6 +205,10 @@
                 <label for="node-config-input-port" style="margin-left:20px; width:43px; "> <span data-i18n="mqtt.label.port"></span></label>
                 <input type="text" id="node-config-input-port" data-i18n="[placeholder]mqtt.label.port" style="width:55px">
             </div>
+            <div class="form-row" style="margin-bottom:0">
+                <input type="checkbox" id="node-config-input-autoConnect" style="margin: 0 5px 0 104px; display: inline-block; width: auto;">
+                <label for="node-config-input-autoConnect" style="width: auto"><span data-i18n="mqtt.label.auto-connect"></span></label>
+            </div>
             <div class="form-row" style="height: 34px;">
                 <input type="checkbox" id="node-config-input-usetls" style="height: 34px; margin: 0 5px 0 104px; display: inline-block; width: auto; vertical-align: top;">
                 <label for="node-config-input-usetls" style="width: 100px; line-height: 34px;"><span data-i18n="mqtt.label.use-tls"></span></label>
@@ -434,6 +458,7 @@
                     return (this.cleansession===undefined || this.cleansession) || (v||"").length > 0;
                 }
             }},
+            autoConnect: {value: true},
             usetls: {value: false},
             verifyservercert: { value: false},
             compatmode: { value: false},
@@ -558,6 +583,10 @@
                 this.usetls = false;
                 $("#node-config-input-usetls").prop("checked",false);
             }
+            if (typeof this.autoConnect === 'undefined') {
+                this.autoConnect = true;
+                $("#node-config-input-autoConnect").prop("checked",true);
+            }
             if (this.compatmode === 'true' || this.compatmode === true) {
                 delete this.compatmode;
                 this.protocolVersion = 4;
@@ -704,11 +733,22 @@
 
         }
     });
+
     RED.nodes.registerType('mqtt in',{
         category: 'network',
         defaults: {
             name: {value:""},
-            topic: {value:"",required:true,validate: RED.validators.regex(/^(#$|(\+|[^+#]*)(\/(\+|[^+#]*))*(\/(\+|#|[^+#]*))?$)/)},
+            topic: {
+                value:"",
+                validate: function(v) {
+                    var isDynamic = this.inputs === 1;
+                    var topicTypeSelect = $("#node-input-topicType");
+                    if (topicTypeSelect.length) {
+                        isDynamic = topicTypeSelect.val()==='dynamic'
+                    }
+                    return isDynamic || ((!!v) && RED.validators.regex(/^(#$|(\+|[^+#]*)(\/(\+|[^+#]*))*(\/(\+|#|[^+#]*))?$)/)(v));
+                }
+            },
             qos: {value: "2"},
             datatype: {value:"auto",required:true},
             broker: {type:"mqtt-broker", required:true},
@@ -716,33 +756,64 @@
             nl: {value:false},
             rap: {value:true},
             rh: {value:0},
+            inputs: {value:0},
         },
         color:"#d8bfd8",
         inputs:0,
         outputs:1,
         icon: "bridge.svg",
         label: function() {
-            return this.name||this.topic||"mqtt";
+            var label = "mqtt";
+            if(this.topicType !== "dynamic" && this.topic) {
+                label = this.topic;
+            }
+            return this.name || label;
         },
         labelStyle: function() {
             return this.name?"node_label_italic":"";
         },
         oneditprepare: function() {
-            $("#node-input-broker").on("change",function(d){
+            const node = this;
+            const isV5Broker = function() {
                 var confNode = RED.nodes.node($("#node-input-broker").val());
-                var v5 = confNode && confNode.protocolVersion == "5";
+                return confNode && confNode.protocolVersion === "5";
-                if(v5) {
+            }
-                    $("div.form-row.mqtt5").show();
+            const isDynamic = function() {
-                } else {
+                return $('#node-input-topicType').val() === "dynamic";
-                    $("div.form-row.mqtt5").hide();
+            }
-                }
+            const updateVisibility = function() {
+                var v5 = isV5Broker();
+                var dynamic = isDynamic();
+                $("div.form-row-mqtt5").toggleClass("form-row-mqtt5-active",!!v5);
+                $("div.form-row.form-row-mqtt-static").toggleClass("form-row-mqtt-static-disabled", !!dynamic)
+            }
+            $("#node-input-broker").on("change",function(d){
+                updateVisibility();
             });
+
+            $('#node-input-topicType').on("change", function () {
+                $("#node-input-inputs").val(isDynamic() ? 1 : 0);
+                updateVisibility();
+            });
+
+            if (this.inputs === 1) {
+                $('#node-input-topicType').val('dynamic')
+            } else {
+                $('#node-input-topicType').val('topic')
+            }
+            $('#node-input-topicType').trigger("change");
+
             if (this.qos === undefined) {
                 $("#node-input-qos").val("2");
             }
             if (this.datatype === undefined) {
                 $("#node-input-datatype").val("auto");
             }
+        },
+        oneditsave: function() {
+            if ($('#node-input-topicType').val() === "dynamic") {
+                $('#node-input-topic').val("");
+            }
         }
     });
 

packages/node_modules/@node-red/nodes/locales/en-US/messages.json

@@ -416,7 +416,11 @@
             "maximumPacketSize": "Max Packet Size",
             "receiveMaximum": "Receive Max",
             "session": "Session",
-            "delay": "Delay"
+            "delay": "Delay",
+            "action": "Action",
+            "staticTopic": "Subscribe to single topic",
+            "dynamicTopic": "Dynamic subscription",
+            "auto-connect": "Connect automatically"
         },
         "sections-label":{
             "birth-message": "Message sent on connection (birth message)",
@@ -457,7 +461,10 @@
             "invalid-topic": "Invalid topic specified",
             "nonclean-missingclientid": "No client ID set, using clean session",
             "invalid-json-string": "Invalid JSON string",
-            "invalid-json-parse": "Failed to parse JSON string"
+            "invalid-json-parse": "Failed to parse JSON string",
+            "invalid-action-action": "Invalid action specified",
+            "invalid-action-alreadyconnected": "Disconnect from broker before connecting",
+            "invalid-action-badsubscription": "msg.topic is missing or invalid"
         }
     },
     "httpin": {

packages/node_modules/@node-red/nodes/locales/en-US/network/10-mqtt.html

@@ -40,6 +40,38 @@
     <p>This node requires a connection to a MQTT broker to be configured. This is configured by clicking
     the pencil icon.</p>
     <p>Several MQTT nodes (in or out) can share the same broker connection if required.</p>
+    <h4>Dynamic Subscription</h4>
+    The node can be configured to dynamically control the MQTT connection and its subscriptions. When
+    enabled, the node will have an input and can be controlled by passing it messages.
+    <h3>Inputs</h3>
+    <p>These only apply when the node has been configured for dynamic subscriptions.</p>
+    <dl class="message-properties">
+       <dt>action <span class="property-type">string</span></dt>
+       <dd>the name of the action the node should perform. Available actions are: <code>"connect"</code>,
+       <code>"disconnect"</code>, <code>"subscribe"</code> and <code>"unsubscribe"</code>.</dd>
+       <dt class="optional">topic <span class="property-type">string|object|array</span></dt>
+       <dd>For the <code>"subscribe"</code> and <code>"unsubscribe"</code> actions, this property
+           provides the topic. It can be set as either:<ul>
+           <li>a String continaing the topic filter</li>
+           <li>an Object containing <code>topic</code> and <code>qos</code> properties</li>
+           <li>an array of either strings or objects to handle multiple topics in one</li>
+            </ul>
+        </dd>
+       <dt class="optional">broker <span class="property-type">broker</span> </dt>
+       <dd>For the <code>"connect"</code> action, this property can override any
+           of the individual broker configuration settings, including: <ul>
+               <li><code>broker</code></li>
+               <li><code>port</code></li>
+               <li><code>url</code> - overrides broker/port to provide a complete connection url</li>
+               <li><code>username</code></li>
+               <li><code>password</code></li>
+           </ul>
+           <p>If this property is set and the broker is already connected an error
+              will be logged unless it has the <code>force</code> property set - in which case it will
+              disconnect from the broker, apply the new settings and reconnect.</p>
+       </dd>
+    </dl>
+
 </script>
 
 <script type="text/html" data-help-name="mqtt out">
@@ -78,6 +110,30 @@
     <p>This node requires a connection to a MQTT broker to be configured. This is configured by clicking
     the pencil icon.</p>
     <p>Several MQTT nodes (in or out) can share the same broker connection if required.</p>
+
+    <h4>Dynamic Control</h4>
+    The connection shared by the node can be controlled dynamically. If the node receives
+    one of the following control messages, it will not publish the message payload as well.
+    <h3>Inputs</h3>
+    <dl class="message-properties">
+       <dt>action <span class="property-type">string</span></dt>
+       <dd>the name of the action the node should perform. Available actions are: <code>"connect"</code>,
+       and <code>"disconnect"</code>.</dd>
+       <dt class="optional">broker <span class="property-type">broker</span> </dt>
+       <dd>For the <code>"connect"</code> action, this property can override any
+           of the individual broker configuration settings, including: <ul>
+               <li><code>broker</code></li>
+               <li><code>port</code></li>
+               <li><code>url</code> - overrides broker/port to provide a complete connection url</li>
+               <li><code>username</code></li>
+               <li><code>password</code></li>
+           </ul>
+           <p>If this property is set and the broker is already connected an error
+              will be logged unless it has the <code>force</code> property set - in which case it will
+              disconnect from the broker, apply the new settings and reconnect.</p>
+       </dd>
+    </dl>
+
 </script>
 
 <script type="text/html" data-help-name="mqtt-broker">