From b8f1386ad0316b2b5abb7fc70d3c53f444321104 Mon Sep 17 00:00:00 2001
From: Nick O'Leary <nick.oleary@gmail.com>
Date: Thu, 14 Oct 2021 12:05:06 +0100
Subject: [PATCH] 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>
---
 .../@node-red/nodes/core/network/10-mqtt.html |  115 +-
 .../@node-red/nodes/core/network/10-mqtt.js   | 1157 ++++++++++-------
 .../nodes/locales/en-US/messages.json         |   11 +-
 .../nodes/locales/en-US/network/10-mqtt.html  |   56 +
 4 files changed, 866 insertions(+), 473 deletions(-)

diff --git a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.html b/packages/node_modules/@node-red/nodes/core/network/10-mqtt.html
index 406abf57d..747281ad0 100644
--- a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.html
+++ b/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">
-        <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">
+    <div class="form-row mqtt-flags-row form-row-mqtt5 form-row-mqtt-static">
         <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";
-                if(v5) {
-                    $("div.form-row.mqtt5").show();
-                } else {
-                    $("div.form-row.mqtt5").hide();
-                }
+                return confNode && confNode.protocolVersion === "5";
+            }
+            const isDynamic = function() {
+                return $('#node-input-topicType').val() === "dynamic";
+            }
+            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("");
+            }
         }
     });
 
diff --git a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js b/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
index 5c1c75881..4e47b17f1 100644
--- a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
+++ b/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
@@ -17,11 +17,11 @@
 module.exports = function(RED) {
     "use strict";
     var mqtt = require("mqtt");
-    var util = require("util");
     var isUtf8 = require('is-utf8');
     var HttpsProxyAgent = require('https-proxy-agent');
     var url = require('url');
 
+    //#region "Supporting functions"
     function matchTopic(ts,t) {
         if (ts == "#") {
             return true;
@@ -37,7 +37,6 @@ module.exports = function(RED) {
         */
         else if(ts.startsWith("$share")){
             ts = ts.replace(/^\$share\/[^#+/]+\/(.*)/g,"$1");
-
         }
         var re = new RegExp("^"+ts.replace(/([\[\]\?\(\)\\\\$\^\*\.|])/g,"\\$1").replace(/\+/g,"[^/]+").replace(/\/#$/,"(\/.*)?")+"$");
         return re.test(t);
@@ -53,7 +52,7 @@ module.exports = function(RED) {
      * @param {integer} [def] An optional default to set in the destination object if prop is NOT present in the soruce object
      */
     function setIntProp(src, dst, propName, minVal, maxVal, def) {
-        if (src.hasOwnProperty(propName)) {
+        if (hasProperty(src, propName)) {
             var v = parseInt(src[propName]);
             if(isNaN(v)) return;
             if(minVal != null) {
@@ -68,6 +67,15 @@ module.exports = function(RED) {
         }
     }
 
+    /**
+     * Test a topic string is valid
+     * @param {string} topic
+     * @returns `true` if it is a valid topic
+     */
+    function isValidSubscriptionTopic(topic) {
+        return /^(#$|(\+|[^+#]*)(\/(\+|[^+#]*))*(\/(\+|#|[^+#]*))?$)/.test(topic)
+    }
+
     /**
      * Helper function for setting string property values in the MQTT V5 properties object
      * @param {object} src Source object containing properties
@@ -146,254 +154,488 @@ module.exports = function(RED) {
         }
     }
 
-    function MQTTBrokerNode(n) {
-        RED.nodes.createNode(this,n);
-
-        // Configuration options passed by Node Red
-        this.broker = n.broker;
-        this.port = n.port;
-        this.clientid = n.clientid;
-        this.usetls = n.usetls;
-        this.usews = n.usews;
-        this.verifyservercert = n.verifyservercert;
-        this.compatmode = n.compatmode;
-        this.protocolVersion = n.protocolVersion;
-        this.keepalive = n.keepalive;
-        this.cleansession = n.cleansession;
-        this.sessionExpiryInterval = n.sessionExpiry;
-        this.topicAliasMaximum = n.topicAliasMaximum;
-        this.maximumPacketSize = n.maximumPacketSize;
-        this.receiveMaximum = n.receiveMaximum;
-        this.userProperties = n.userProperties;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901116
-        this.userPropertiesType = n.userPropertiesType;
-
-        // Config node state
-        this.brokerurl = "";
-        this.connected = false;
-        this.connecting = false;
-        this.closing = false;
-        this.options = {};
-        this.queue = [];
-        this.subscriptions = {};
-
-        if (n.birthTopic) {
-            this.birthMessage = {
-                topic: n.birthTopic,
-                payload: n.birthPayload || "",
-                qos: Number(n.birthQos||0),
-                retain: n.birthRetain=="true"|| n.birthRetain===true,
-                //TODO: add payloadFormatIndicator, messageExpiryInterval, contentType, responseTopic, correlationData, userProperties
-            };
-            if (n.birthMsg) {
-                setStrProp(n.birthMsg, this.birthMessage, "contentType");
-                if(n.birthMsg.userProps && /^ *{/.test(n.birthMsg.userProps)) {
-                    try {
-                        setUserProperties(JSON.parse(n.birthMsg.userProps), this.birthMessage);
-                    } catch(err) {}
-                }
-                n.birthMsg.responseTopic = n.birthMsg.respTopic;
-                setStrProp(n.birthMsg, this.birthMessage, "responseTopic");
-                n.birthMsg.correlationData = n.birthMsg.correl;
-                setBufferProp(n.birthMsg, this.birthMessage, "correlationData");
-                n.birthMsg.messageExpiryInterval = n.birthMsg.expiry
-                setIntProp(n.birthMsg,this.birthMessage, "messageExpiryInterval")
+    /**
+     * Helper function for applying changes to an objects properties ONLY when the src object actually has the property.
+     * This avoids setting a `dst` property null/undefined when the `src` object doesnt have the named property.
+     * @param {object} src Source object containing properties
+     * @param {object} dst Destination object to set property
+     * @param {string} propName The property name to set in the Destination object
+     * @param {boolean} force force the dst property to be updated/created even if src property is empty
+     */
+    function setIfHasProperty(src, dst, propName, force) {
+        if (src && dst && propName) {
+            const ok = force || hasProperty(src, propName);
+            if (ok) {
+                dst[propName] = src[propName];
             }
         }
+    }
 
-        if (n.closeTopic) {
-            this.closeMessage = {
-                topic: n.closeTopic,
-                payload: n.closePayload || "",
-                qos: Number(n.closeQos||0),
-                retain: n.closeRetain=="true"|| n.closeRetain===true,
-                //TODO: add payloadFormatIndicator, messageExpiryInterval, contentType, responseTopic, correlationData, userProperties
-            };
-            if (n.closeMsg) {
-                setStrProp(n.closeMsg, this.closeMessage, "contentType");
-                if(n.closeMsg.userProps && /^ *{/.test(n.closeMsg.userProps)) {
-                    try {
-                        setUserProperties(JSON.parse(n.closeMsg.userProps), this.closeMessage);
-                    } catch(err) {}
-                }
-                n.closeMsg.responseTopic = n.closeMsg.respTopic;
-                setStrProp(n.closeMsg, this.closeMessage, "responseTopic");
-                n.closeMsg.correlationData = n.closeMsg.correl;
-                setBufferProp(n.closeMsg, this.closeMessage, "correlationData");
-                n.closeMsg.messageExpiryInterval = n.closeMsg.expiry
-                setIntProp(n.birthMsg,this.closeMessage, "messageExpiryInterval")
+    /**
+     * Helper function to test an object has a property
+     * @param {object} obj Object to test
+     * @param {string} propName Name of property to find
+     * @returns true if object has property `propName`
+     */
+    function hasProperty(obj, propName) {
+        //JavaScript does not protect the property name hasOwnProperty
+        //Object.prototype.hasOwnProperty.call is the recommended/safer test
+        return Object.prototype.hasOwnProperty.call(obj, propName);
+    }
+
+    /**
+     * Handle the payload / packet recieved in MQTT In and MQTT Sub nodes
+     */
+    function subscriptionHandler(node, datatype ,topic, payload, packet) {
+        const v5 = node.brokerConn.options && node.brokerConn.options.protocolVersion == 5;
+
+        if (datatype === "buffer") {
+            // payload = payload;
+        } else if (datatype === "base64") {
+            payload = payload.toString('base64');
+        } else if (datatype === "utf8") {
+            payload = payload.toString('utf8');
+        } else if (datatype === "json") {
+            if (isUtf8(payload)) {
+                payload = payload.toString();
+                try { payload = JSON.parse(payload); }
+                catch(e) { node.error(RED._("mqtt.errors.invalid-json-parse"),{payload:payload, topic:topic, qos:packet.qos, retain:packet.retain}); return; }
             }
+            else { node.error((RED._("mqtt.errors.invalid-json-string")),{payload:payload, topic:topic, qos:packet.qos, retain:packet.retain}); return; }
+        } else {
+            if (isUtf8(payload)) { payload = payload.toString(); }
         }
+        var msg = {topic:topic, payload:payload, qos:packet.qos, retain:packet.retain};
+        if(v5 && packet.properties) {
+            setStrProp(packet.properties, msg, "responseTopic");
+            setBufferProp(packet.properties, msg, "correlationData");
+            setStrProp(packet.properties, msg, "contentType");
+            setIntProp(packet.properties, msg, "messageExpiryInterval", 0);
+            setBoolProp(packet.properties, msg, "payloadFormatIndicator");
+            setStrProp(packet.properties, msg, "reasonString");
+            setUserProperties(packet.properties.userProperties, msg);
+        }
+        if ((node.brokerConn.broker === "localhost")||(node.brokerConn.broker === "127.0.0.1")) {
+            msg._topic = topic;
+        }
+        node.send(msg);
+    }
 
-        if (this.credentials) {
-            this.username = this.credentials.user;
-            this.password = this.credentials.password;
-        }
+    /**
+     * Send an mqtt message to broker
+     * @param {MQTTOutNode} node the owner node
+     * @param {object} msg The msg to prepare for publishing
+     * @param {function} done callback when done
+     */
+    function doPublish(node, msg, done) {
+        try {
+            done = typeof done == "function" ? done : function noop(){};
+            let v5 = node.brokerConn.options && node.brokerConn.options.protocolVersion == 5;
+            const bsp = (node.brokerConn && node.brokerConn.serverProperties) || {};
 
-        // If the config node is missing certain options (it was probably deployed prior to an update to the node code),
-        // select/generate sensible options for the new fields
-        if (typeof this.usetls === 'undefined') {
-            this.usetls = false;
-        }
-        if (typeof this.usews === 'undefined') {
-            this.usews = false;
-        }
-        if (typeof this.verifyservercert === 'undefined') {
-            this.verifyservercert = false;
-        }
-        if (typeof this.keepalive === 'undefined') {
-            this.keepalive = 60;
-        } else if (typeof this.keepalive === 'string') {
-            this.keepalive = Number(this.keepalive);
-        }
-        if (typeof this.cleansession === 'undefined') {
-            this.cleansession = true;
-        }
+            //Sanitise the `msg` object properties ready for publishing
+            if (msg.qos) {
+                msg.qos = parseInt(msg.qos);
+                if ((msg.qos !== 0) && (msg.qos !== 1) && (msg.qos !== 2)) {
+                    msg.qos = null;
+                }
+            }
 
-        var prox, noprox;
-        if (process.env.http_proxy) { prox = process.env.http_proxy; }
-        if (process.env.HTTP_PROXY) { prox = process.env.HTTP_PROXY; }
-        if (process.env.no_proxy) { noprox = process.env.no_proxy.split(","); }
-        if (process.env.NO_PROXY) { noprox = process.env.NO_PROXY.split(","); }
+            /* If node properties exists, override/set that to property in msg  */
+            if (node.topic) { msg.topic = node.topic; }
+            msg.qos = Number(node.qos || msg.qos || 0);
+            msg.retain = node.retain || msg.retain || false;
+            msg.retain = ((msg.retain === true) || (msg.retain === "true")) || false;
 
+            if (v5) {
+                if (node.userProperties) {
+                    msg.userProperties = node.userProperties;
+                }
+                if (node.responseTopic) {
+                    msg.responseTopic = node.responseTopic;
+                }
+                if (node.correlationData) {
+                    msg.correlationData = node.correlationData;
+                }
+                if (node.contentType) {
+                    msg.contentType = node.contentType;
+                }
+                if (node.messageExpiryInterval) {
+                    msg.messageExpiryInterval = node.messageExpiryInterval;
+                }
+            }
+            if (msg.userProperties && typeof msg.userProperties !== "object") {
+                delete msg.userProperties;
+            }
+            if (hasProperty(msg, "topicAlias") && !isNaN(msg.topicAlias) && (msg.topicAlias === 0 || bsp.topicAliasMaximum === 0 || msg.topicAlias > bsp.topicAliasMaximum)) {
+                delete msg.topicAlias;
+            }
 
-        // Create the URL to pass in to the MQTT.js library
-        if (this.brokerurl === "") {
-            // if the broker may be ws:// or wss:// or even tcp://
-            if (this.broker.indexOf("://") > -1) {
-                this.brokerurl = this.broker;
-                // Only for ws or wss, check if proxy env var for additional configuration
-                if (this.brokerurl.indexOf("wss://") > -1 || this.brokerurl.indexOf("ws://") > -1 ) {
-                    // check if proxy is set in env
-                    var noproxy;
-                    if (noprox) {
-                        for (var i = 0; i < noprox.length; i += 1) {
-                            if (this.brokerurl.indexOf(noprox[i].trim()) !== -1) { noproxy=true; }
-                        }
-                    }
-                    if (prox && !noproxy) {
-                        var parsedUrl = url.parse(this.brokerurl);
-                        var proxyOpts = url.parse(prox);
-                        // true for wss
-                        proxyOpts.secureEndpoint = parsedUrl.protocol ? parsedUrl.protocol === 'wss:' : true;
-                        // Set Agent for wsOption in MQTT
-                        var agent = new HttpsProxyAgent(proxyOpts);
-                        this.options.wsOptions = {
-                            agent: agent
-                        }
+            if (hasProperty(msg, "payload")) {
+
+                //check & sanitise topic
+                let topicOK = hasProperty(msg, "topic") && (typeof msg.topic === "string") && (msg.topic !== "");
+
+                if (!topicOK && v5) {
+                    //NOTE: A value of 0 (in server props topicAliasMaximum) indicates that the Server does not accept any Topic Aliases on this connection
+                    if (hasProperty(msg, "topicAlias") && !isNaN(msg.topicAlias) && msg.topicAlias >= 0 && bsp.topicAliasMaximum && bsp.topicAliasMaximum >= msg.topicAlias) {
+                        topicOK = true;
+                        msg.topic = ""; //must be empty string
+                    } else if (hasProperty(msg, "responseTopic") && (typeof msg.responseTopic === "string") && (msg.responseTopic !== "")) {
+                        //TODO: if topic is empty but responseTopic has a string value, use that instead. Is this desirable?
+                        topicOK = true;
+                        msg.topic = msg.responseTopic;
+                        //TODO: delete msg.responseTopic - to prevent it being resent?
                     }
                 }
+                topicOK = topicOK && !/[\+#]/.test(msg.topic);
+
+                if (topicOK) {
+                    node.brokerConn.publish(msg, done); // send the message
+                } else {
+                    node.warn(RED._("mqtt.errors.invalid-topic"));
+                    done();
+                }
             } else {
-                // construct the std mqtt:// url
-                if (this.usetls) {
-                    this.brokerurl="mqtts://";
-                } else {
-                    this.brokerurl="mqtt://";
-                }
-                if (this.broker !== "") {
-                    //Check for an IPv6 address
-                    if (/(?:^|(?<=\s))(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))(?=\s|$)/.test(this.broker)) {
-                        this.brokerurl = this.brokerurl+"["+this.broker+"]:";
-                    } else {
-                        this.brokerurl = this.brokerurl+this.broker+":";
-                    }
-                    // port now defaults to 1883 if unset.
-                    if (!this.port){
-                        this.brokerurl = this.brokerurl+"1883";
-                    } else {
-                        this.brokerurl = this.brokerurl+this.port;
-                    }
-                } else {
-                    this.brokerurl = this.brokerurl+"localhost:1883";
+                done();
+            }
+        } catch (error) {
+            done(error);
+        }
+    }
+
+    function setStatusDisconnected(node, allNodes) {
+        if(allNodes) {
+            for (var id in node.users) {
+                if (hasProperty(node.users, id)) {
+                    node.users[id].status({ fill: "red", shape: "ring", text: "node-red:common.status.disconnected" });
                 }
             }
+        } else {
+            node.status({ fill: "red", shape: "ring", text: "node-red:common.status.disconnected" });
         }
+    }
 
-        if (!this.cleansession && !this.clientid) {
-            this.cleansession = true;
-            this.warn(RED._("mqtt.errors.nonclean-missingclientid"));
-        }
-
-        // Build options for passing to the MQTT.js API
-        this.options.clientId = this.clientid || 'mqtt_' + RED.util.generateId();
-        this.options.username = this.username;
-        this.options.password = this.password;
-        this.options.keepalive = this.keepalive;
-        this.options.clean = this.cleansession;
-        this.options.reconnectPeriod = RED.settings.mqttReconnectTime||5000;
-        if (this.compatmode == "true" || this.compatmode === true || this.protocolVersion == 3) {
-            this.options.protocolId = 'MQIsdp';
-            this.options.protocolVersion = 3;
-        } else if ( this.protocolVersion == 5 ) {
-            this.options.protocolVersion = 5;
-            this.options.properties = {};
-            this.options.properties.requestResponseInformation = true;
-            this.options.properties.requestProblemInformation = true;
-            if(this.userProperties && /^ *{/.test(this.userProperties)) {
-                try {
-                    setUserProperties(JSON.parse(this.userProperties), this.options.properties);
-                } catch(err) {}
+    function setStatusConnecting(node, allNodes) {
+        if(allNodes) {
+            for (var id in node.users) {
+                if (hasProperty(node.users, id)) {
+                    node.users[id].status({ fill: "yellow", shape: "ring", text: "node-red:common.status.connecting" });
+                }
             }
-            if (this.sessionExpiryInterval && this.sessionExpiryInterval !== "0") {
-                setIntProp(this,this.options.properties,"sessionExpiryInterval");
+        } else {
+            node.status({ fill: "yellow", shape: "ring", text: "node-red:common.status.connecting" });
+        }
+    }
+
+    function setStatusConnected(node, allNodes) {
+        if(allNodes) {
+            for (var id in node.users) {
+                if (hasProperty(node.users, id)) {
+                    node.users[id].status({ fill: "green", shape: "dot", text: "node-red:common.status.connected" });
+                }
+            }
+        } else {
+            node.status({ fill: "green", shape: "dot", text: "node-red:common.status.connected" });
+        }
+    }
+
+    function handleConnectAction(node, msg, done) {
+        let actionData = typeof msg.broker === 'object' ? msg.broker : null;
+        if (node.brokerConn.canConnect()) {
+            // Not currently connected/connecting - trigger the connect
+            if (actionData) {
+                node.brokerConn.setOptions(actionData);
+            }
+            node.brokerConn.connect(function () {
+                done();
+            });
+        } else {
+            // Already Connected/Connecting
+            if (!actionData) {
+                // All is good - already connected and no broker override provided
+                done()
+            } else if (actionData.force) {
+                // The force flag tells us to cycle the connection.
+                node.brokerConn.disconnect(function() {
+                    node.brokerConn.setOptions(actionData);
+                    node.brokerConn.connect(function () {
+                        done();
+                    });
+                })
+            } else {
+                // Without force flag, we will refuse to cycle an active connection
+                done(new Error(RED._('mqtt.errors.invalid-action-alreadyconnected')));
             }
         }
-        if (this.usetls && n.tls) {
-            var tlsNode = RED.nodes.getNode(n.tls);
-            if (tlsNode) {
-                tlsNode.addTLSOptions(this.options);
+    }
+
+    function handleDisconnectAction(node, done) {
+        node.brokerConn.disconnect(function () {
+            done();
+        });
+    }
+
+    //#endregion  "Supporting functions"
+
+    //#region  "Broker node"
+    function MQTTBrokerNode(n) {
+        RED.nodes.createNode(this,n);
+        const node = this;
+        node.users = {};
+        // Config node state
+        node.brokerurl = "";
+        node.connected = false;
+        node.connecting = false;
+        node.closing = false;
+        node.options = {};
+        node.queue = [];
+        node.subscriptions = {};
+        /** @type {mqtt.MqttClient}*/ this.client;
+        node.setOptions = function(opts, init) {
+            if(!opts || typeof opts !== "object") {
+                return; //nothing to change, simply return
             }
-        }
+            const originalBrokerURL = node.brokerurl;
 
-        // If there's no rejectUnauthorized already, then this could be an
-        // old config where this option was provided on the broker node and
-        // not the tls node
-        if (typeof this.options.rejectUnauthorized === 'undefined') {
-            this.options.rejectUnauthorized = (this.verifyservercert == "true" || this.verifyservercert === true);
-        }
+            //apply property changes (only if the property exists in the opts object)
+            setIfHasProperty(opts, node, "url", init);
+            setIfHasProperty(opts, node, "broker", init);
+            setIfHasProperty(opts, node, "port", init);
+            setIfHasProperty(opts, node, "clientid", init);
+            setIfHasProperty(opts, node, "autoConnect", init);
+            setIfHasProperty(opts, node, "usetls", init);
+            setIfHasProperty(opts, node, "usews", init);
+            setIfHasProperty(opts, node, "verifyservercert", init);
+            setIfHasProperty(opts, node, "compatmode", init);
+            setIfHasProperty(opts, node, "protocolVersion", init);
+            setIfHasProperty(opts, node, "keepalive", init);
+            setIfHasProperty(opts, node, "cleansession", init);
+            setIfHasProperty(opts, node, "sessionExpiry", init);
+            setIfHasProperty(opts, node, "topicAliasMaximum", init);
+            setIfHasProperty(opts, node, "maximumPacketSize", init);
+            setIfHasProperty(opts, node, "receiveMaximum", init);
+            setIfHasProperty(opts, node, "userProperties", init);//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901116
+            setIfHasProperty(opts, node, "userPropertiesType", init);
 
-        if (n.willTopic) {
-            this.options.will = {
-                topic: n.willTopic,
-                payload: n.willPayload || "",
-                qos: Number(n.willQos||0),
-                retain: n.willRetain=="true"|| n.willRetain===true,
-                //TODO: add willDelayInterval, payloadFormatIndicator, messageExpiryInterval, contentType, responseTopic, correlationData, userProperties
+            function createLWT(topic, payload, qos, retain, v5opts, v5SubPropName) {
+                let message = undefined;
+                if(topic) {
+                    message = {
+                        topic: topic,
+                        payload: payload || "",
+                        qos: Number(qos||0),
+                        retain: retain=="true"|| retain===true,
+                    }
+                    if (v5opts) {
+                        let v5Properties = message;
+                        if(v5SubPropName) {
+                            v5Properties = message[v5SubPropName] = {};
+                        }
+                        //re-align local prop name to mqttjs std
+                        if(hasProperty(v5opts, "respTopic")) { v5opts.responseTopic = v5opts.respTopic; }
+                        if(hasProperty(v5opts, "correl")) { v5opts.correlationData = v5opts.correl; }
+                        if(hasProperty(v5opts, "expiry")) { v5opts.messageExpiryInterval = v5opts.expiry; }
+                        if(hasProperty(v5opts, "delay")) { v5opts.willDelayInterval = v5opts.delay; }
+                        if(hasProperty(v5opts, "userProps")) { v5opts.userProperties = v5opts.userProps; }
+                        //setup v5 properties
+                        if(typeof v5opts.userProperties == "string" && /^ *{/.test(v5opts.userProperties)) {
+                            try {
+                                setUserProperties(JSON.parse(v5opts.userProps), v5Properties);
+                            } catch(err) {}
+                        } else if (typeof v5opts.userProperties == "object") {
+                            setUserProperties(v5opts.userProperties, v5Properties);
+                        }
+                        setStrProp(v5opts, v5Properties, "contentType");
+                        setStrProp(v5opts, v5Properties, "responseTopic");
+                        setBufferProp(v5opts, v5Properties, "correlationData");
+                        setIntProp(v5opts, v5Properties, "messageExpiryInterval");
+                        setIntProp(v5opts, v5Properties, "willDelayInterval");
+                    }
+                }
+                return message;
+            }
+
+            if(init) {
+                if(hasProperty(opts, "birthTopic")) {
+                    node.birthMessage = createLWT(opts.birthTopic, opts.birthPayload, opts.birthQos, opts.birthRetain, opts.birthMsg, "");
+                };
+                if(hasProperty(opts, "closeTopic")) {
+                    node.closeMessage = createLWT(opts.closeTopic, opts.closePayload, opts.closeQos, opts.closeRetain, opts.closeMsg, "");
+                };
+                if(hasProperty(opts, "willTopic")) {
+                    //will v5 properties must be set in the "properties" sub object
+                    node.options.will = createLWT(opts.willTopic, opts.willPayload, opts.willQos, opts.willRetain, opts.willMsg, "properies");
+                };
+            } else {
+                //update options
+                if(hasProperty(opts, "birth")) {
+                    if(typeof opts.birth !== "object") { opts.birth = {}; }
+                    node.birthMessage = createLWT(opts.birth.topic, opts.birth.payload, opts.birth.qos, opts.birth.retain, opts.birth.properties, "");
+                }
+                if(hasProperty(opts, "close")) {
+                    if(typeof opts.close !== "object") { opts.close = {}; }
+                    node.closeMessage = createLWT(opts.close.topic, opts.close.payload, opts.close.qos, opts.close.retain, opts.close.properties, "");
+                }
+                if(hasProperty(opts, "will")) {
+                    if(typeof opts.will !== "object") { opts.will = {}; }
+                    //will v5 properties must be set in the "properties" sub object
+                    node.options.will = createLWT(opts.will.topic, opts.will.payload, opts.will.qos, opts.will.retain, opts.will.properties, "properties");
+                }
+            }
+
+            if (node.credentials) {
+                node.username = node.credentials.user;
+                node.password = node.credentials.password;
+            }
+            if(!init & hasProperty(opts, "username")) {
+                node.username  = opts.username;
+            };
+            if(!init & hasProperty(opts, "password")) {
+                node.password  = opts.password;
             };
-            if (n.willMsg) {
-                this.options.will.properties = {};
 
-                setStrProp(n.willMsg, this.options.will.properties, "contentType");
-                if(n.willMsg.userProps && /^ *{/.test(n.willMsg.userProps)) {
+            // If the config node is missing certain options (it was probably deployed prior to an update to the node code),
+            // select/generate sensible options for the new fields
+            if (typeof node.usetls === 'undefined') {
+                node.usetls = false;
+            }
+            if (typeof node.usews === 'undefined') {
+                node.usews = false;
+            }
+            if (typeof node.verifyservercert === 'undefined') {
+                node.verifyservercert = false;
+            }
+            if (typeof node.keepalive === 'undefined') {
+                node.keepalive = 60;
+            } else if (typeof node.keepalive === 'string') {
+                node.keepalive = Number(node.keepalive);
+            }
+            if (typeof node.cleansession === 'undefined') {
+                node.cleansession = true;
+            }
+
+            //use url or build a url from usetls://broker:port
+            if (node.url && node.brokerurl !== node.url) {
+                node.brokerurl = node.url;
+            } else {
+                // if the broker is ws:// or wss:// or tcp://
+                if (node.broker.indexOf("://") > -1) {
+                    node.brokerurl = node.broker;
+                    // Only for ws or wss, check if proxy env var for additional configuration
+                    if (node.brokerurl.indexOf("wss://") > -1 || node.brokerurl.indexOf("ws://") > -1) {
+                        // check if proxy is set in env
+                        let prox, noprox;
+                        if (process.env.http_proxy) { prox = process.env.http_proxy; }
+                        if (process.env.HTTP_PROXY) { prox = process.env.HTTP_PROXY; }
+                        if (process.env.no_proxy) { noprox = process.env.no_proxy.split(","); }
+                        if (process.env.NO_PROXY) { noprox = process.env.NO_PROXY.split(","); }
+                        if (noprox) {
+                            for (var i = 0; i < noprox.length; i += 1) {
+                                if (node.brokerurl.indexOf(noprox[i].trim()) !== -1) { noproxy = true; }
+                            }
+                        }
+                        if (prox && !noproxy) {
+                            var parsedUrl = url.parse(node.brokerurl);
+                            var proxyOpts = url.parse(prox);
+                            // true for wss
+                            proxyOpts.secureEndpoint = parsedUrl.protocol ? parsedUrl.protocol === 'wss:' : true;
+                            // Set Agent for wsOption in MQTT
+                            var agent = new HttpsProxyAgent(proxyOpts);
+                            node.options.wsOptions = {
+                                agent: agent
+                            };
+                        }
+                    }
+                } else {
+                    // construct the std mqtt:// url
+                    if (node.usetls) {
+                        node.brokerurl = "mqtts://";
+                    } else {
+                        node.brokerurl = "mqtt://";
+                    }
+                    if (node.broker !== "") {
+                        //Check for an IPv6 address
+                        if (/(?:^|(?<=\s))(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))(?=\s|$)/.test(node.broker)) {
+                            node.brokerurl = node.brokerurl + "[" + node.broker + "]:";
+                        } else {
+                            node.brokerurl = node.brokerurl + node.broker + ":";
+                        }
+                        // port now defaults to 1883 if unset.
+                        if (!node.port) {
+                            node.brokerurl = node.brokerurl + "1883";
+                        } else {
+                            node.brokerurl = node.brokerurl + node.port;
+                        }
+                    } else {
+                        node.brokerurl = node.brokerurl + "localhost:1883";
+                    }
+                }
+            }
+
+            // Ensure cleansession set if clientid not supplied
+            if (!node.cleansession && !node.clientid) {
+                node.cleansession = true;
+                node.warn(RED._("mqtt.errors.nonclean-missingclientid"));
+            }
+
+            // Build options for passing to the MQTT.js API
+            node.options.username = node.username;
+            node.options.password = node.password;
+            node.options.keepalive = node.keepalive;
+            node.options.clean = node.cleansession;
+            node.options.clientId = node.clientid || 'nodered_' + RED.util.generateId();
+            node.options.reconnectPeriod = RED.settings.mqttReconnectTime||5000;
+            delete node.options.protocolId; //V4+ default
+            delete node.options.protocolVersion; //V4 default
+            delete node.options.properties;//V5 only
+            if (node.compatmode == "true" || node.compatmode === true || node.protocolVersion == 3) {
+                node.options.protocolId = 'MQIsdp';//V3 compat only
+                node.options.protocolVersion = 3;
+            } else if ( node.protocolVersion == 5 ) {
+                delete node.options.protocolId;
+                node.options.protocolVersion = 5;
+                node.options.properties = {};
+                node.options.properties.requestResponseInformation = true;
+                node.options.properties.requestProblemInformation = true;
+                if(node.userProperties && /^ *{/.test(node.userProperties)) {
                     try {
-                        setUserProperties(JSON.parse(n.willMsg.userProps), this.options.will.properties);
+                        setUserProperties(JSON.parse(node.userProperties), node.options.properties);
                     } catch(err) {}
                 }
-                n.willMsg.responseTopic = n.willMsg.respTopic;
-                setStrProp(n.willMsg, this.options.will.properties, "responseTopic");
-                n.willMsg.correlationData = n.willMsg.correl;
-                setBufferProp(n.willMsg, this.options.will.properties, "correlationData");
-                n.willMsg.willDelayInterval = n.willMsg.delay
-                setIntProp(n.willMsg,this.options.will.properties, "willDelayInterval")
-                n.willMsg.messageExpiryInterval = n.willMsg.expiry
-                setIntProp(n.willMsg,this.options.will.properties, "messageExpiryInterval")
-                this.options.will.payloadFormatIndicator = true;
+                if (node.sessionExpiryInterval && node.sessionExpiryInterval !== "0") {
+                    setIntProp(node,node.options.properties,"sessionExpiryInterval");
+                }
+            }
+            if (node.usetls && n.tls) {
+                var tlsNode = RED.nodes.getNode(n.tls);
+                if (tlsNode) {
+                    tlsNode.addTLSOptions(node.options);
+                }
+            }
+
+            // If there's no rejectUnauthorized already, then this could be an
+            // old config where this option was provided on the broker node and
+            // not the tls node
+            if (typeof node.options.rejectUnauthorized === 'undefined') {
+                node.options.rejectUnauthorized = (node.verifyservercert == "true" || node.verifyservercert === true);
             }
         }
 
-        // console.log(this.brokerurl,this.options);
+        n.autoConnect = n.autoConnect === "false" || n.autoConnect === false ? false : true;
+        node.setOptions(n, true);
 
         // Define functions called by MQTT in and out nodes
-        var node = this;
-        this.users = {};
-
-        this.register = function(mqttNode) {
+        node.register = function(mqttNode) {
             node.users[mqttNode.id] = mqttNode;
             if (Object.keys(node.users).length === 1) {
-                node.connect();
+                if(node.autoConnect) {
+                    node.connect();
+                }
             }
         };
 
-        this.deregister = function(mqttNode,done) {
+        node.deregister = function(mqttNode,done) {
             delete node.users[mqttNode.id];
             if (node.closing) {
                 return done();
@@ -410,27 +652,35 @@ module.exports = function(RED) {
                     }
                     return;
                 } else {
-                    node.client.end();
+                    if (node.client) { node.client.end(); }
                     return done();
                 }
             }
             done();
         };
-
-        this.connect = function () {
-            if (!node.connected && !node.connecting) {
+        node.canConnect = function() {
+            return !node.connected && !node.connecting;
+        }
+        node.connect = function (callback) {
+            if (node.canConnect()) {
                 node.connecting = true;
+                setStatusConnecting(node, true);
                 try {
                     node.serverProperties = {};
-                    node.client = mqtt.connect(node.brokerurl ,node.options);
+                    node.client = mqtt.connect(node.brokerurl, node.options);
                     node.client.setMaxListeners(0);
+                    let callbackDone = false; //prevent re-connects causing node.client.on('connect' firing callback multiple times
                     // Register successful connect or reconnect handler
                     node.client.on('connect', function (connack) {
                         node.connecting = false;
                         node.connected = true;
+                        if(!callbackDone && typeof callback == "function") {
+                            callback();
+                        }
+                        callbackDone = true;
                         node.topicAliases = {};
                         node.log(RED._("mqtt.state.connected",{broker:(node.clientid?node.clientid+"@":"")+node.brokerurl}));
-                        if(node.options.protocolVersion == 5 && connack && connack.hasOwnProperty("properties")) {
+                        if(node.options.protocolVersion == 5 && connack && hasProperty(connack, "properties")) {
                             if(typeof connack.properties == "object") {
                                 //clean & assign all props sent from server.
                                 setIntProp(connack.properties, node.serverProperties, "topicAliasMaximum", 0);
@@ -448,14 +698,9 @@ module.exports = function(RED) {
                                 setStrProp(connack.properties, node.serverProperties, "assignedClientIdentifier");
                                 setStrProp(connack.properties, node.serverProperties, "reasonString");
                                 setUserProperties(connack.properties, node.serverProperties);
-                                // node.debug("CONNECTED. node.serverProperties ==> "+JSON.stringify(node.serverProperties));//TODO: remove
-                            }
-                        }
-                        for (var id in node.users) {
-                            if (node.users.hasOwnProperty(id)) {
-                                node.users[id].status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
                             }
                         }
+                        setStatusConnected(node, true);
                         // Remove any existing listeners before resubscribing to avoid duplicates in the event of a re-connection
                         node.client.removeAllListeners('message');
 
@@ -483,11 +728,7 @@ module.exports = function(RED) {
                         }
                     });
                     node.client.on("reconnect", function() {
-                        for (var id in node.users) {
-                            if (node.users.hasOwnProperty(id)) {
-                                node.users[id].status({fill:"yellow",shape:"ring",text:"node-red:common.status.connecting"});
-                            }
-                        }
+                        setStatusConnecting(node, true);
                     });
                     //TODO: what to do with this event? Anything? Necessary?
                     node.client.on("disconnect", function(packet) {
@@ -501,11 +742,7 @@ module.exports = function(RED) {
                         if (node.connected) {
                             node.connected = false;
                             node.log(RED._("mqtt.state.disconnected",{broker:(node.clientid?node.clientid+"@":"")+node.brokerurl}));
-                            for (var id in node.users) {
-                                if (node.users.hasOwnProperty(id)) {
-                                    node.users[id].status({fill:"red",shape:"ring",text:"node-red:common.status.disconnected"});
-                                }
-                            }
+                            setStatusDisconnected(node, true);
                         } else if (node.connecting) {
                             node.log(RED._("mqtt.state.connect-failed",{broker:(node.clientid?node.clientid+"@":"")+node.brokerurl}));
                         }
@@ -520,10 +757,31 @@ module.exports = function(RED) {
                 }
             }
         };
+        node.disconnect = function (callback) {
+            const _callback = function () {
+                setStatusDisconnected(node, true);
+                node.connecting = false;
+                node.connected = false;
+                callback && typeof callback == "function" && callback();
+            };
 
-        this.subscriptionIds = {};
-        this.subid = 1;
-        this.subscribe = function (topic,options,callback,ref) {
+            if(node.client) {
+                if(node.client.connected && node.closeMessage) {
+                    node.publish(node.closeMessage, function (err) {
+                        node.client.end(_callback);
+                    });
+                } else if(node.client.connected || node.client.reconnecting) {
+                    node.client.end(_callback);
+                } else if(node.client.disconnecting || node.client.connected === false) {
+                    _callback();
+                }
+            } else {
+                _callback();
+            }
+        }
+        node.subscriptionIds = {};
+        node.subid = 1;
+        node.subscribe = function (topic,options,callback,ref) {
             ref = ref||0;
             var qos;
             if(typeof options == "object") {
@@ -547,58 +805,45 @@ module.exports = function(RED) {
                 handler:function(mtopic,mpayload, mpacket) {
                     if(mpacket.properties && options.properties && mpacket.properties.subscriptionIdentifier && options.properties.subscriptionIdentifier && (mpacket.properties.subscriptionIdentifier !== options.properties.subscriptionIdentifier) ) {
                         //do nothing as subscriptionIdentifier does not match
-                        // node.debug(`> no match - this nodes subID (${options.properties.subscriptionIdentifier}) !== packet subID (${mpacket.properties.subscriptionIdentifier})`); //TODO: remove
                     } else if (matchTopic(topic,mtopic)) {
-                        // node.debug(`>  MATCHED '${topic}' to '${mtopic}' - performing callback`); //TODO: remove
                         callback(mtopic,mpayload, mpacket);
-                    } else {
-                        // node.debug(`> no match / no callback`); //TODO: remove
                     }
                 },
                 ref: ref
             };
             node.subscriptions[topic][ref] = sub;
             if (node.connected) {
-                // node.debug(`this.subscribe - registering handler ref ${ref} for ${topic} and subscribing `+JSON.stringify(options)); //TODO: remove
                 node.client.on('message',sub.handler);
                 node.client.subscribe(topic, options);
             }
         };
 
-        this.unsubscribe = function (topic, ref, removed) {
+        node.unsubscribe = function (topic, ref, removed) {
             ref = ref||0;
             var sub = node.subscriptions[topic];
-            // var _debug = `unsubscribe for topic ${topic} called... ` ; //TODO: remove
             if (sub) {
-                // _debug += "sub found. " //TODO: remove
                 if (sub[ref]) {
-                    // debug(`this.unsubscribe - removing handler ref ${ref} for ${topic} `); //TODO: remove
-                    // _debug += `removing handler ref ${ref} for ${topic}. `
-                    node.client.removeListener('message',sub[ref].handler);
+                    if(node.client) {
+                        node.client.removeListener('message',sub[ref].handler);
+                    }
                     delete sub[ref];
                 }
                 //TODO: Review. The `if(removed)` was commented out to always delete and remove subscriptions.
                 // if we dont then property changes dont get applied and old subs still trigger
                 //if (removed) {
-
                     if (Object.keys(sub).length === 0) {
                         delete node.subscriptions[topic];
                         delete node.subscriptionIds[topic];
                         if (node.connected) {
-                            // _debug += `calling client.unsubscribe to remove topic ${topic}` //TODO: remove
                             node.client.unsubscribe(topic);
                         }
                     }
                 //}
-            } else {
-                // _debug += "sub not found! "; //TODO: remove
             }
-            // node.debug(_debug); //TODO: remove
-
         };
-        this.topicAliases = {};
+        node.topicAliases = {};
 
-        this.publish = function (msg,done) {
+        node.publish = function (msg,done) {
             if (node.connected) {
                 if (msg.payload === null || msg.payload === undefined) {
                     msg.payload = "";
@@ -644,19 +889,9 @@ module.exports = function(RED) {
             }
         };
 
-        this.on('close', function(done) {
-            this.closing = true;
-            if (this.connected) {
-                this.client.once('close', function() {
-                    done();
-                });
-                this.client.end();
-            } else if (this.connecting || node.client.reconnecting) {
-                node.client.end();
-                done();
-            } else {
-                done();
-            }
+        node.on('close', function(done) {
+            node.closing = true;
+            node.disconnect(done);
         });
 
     }
@@ -667,214 +902,238 @@ module.exports = function(RED) {
             password: {type: "password"}
         }
     });
+    //#endregion  "Broker node"
 
+    //#region  "MQTTIn node"
     function MQTTInNode(n) {
         RED.nodes.createNode(this,n);
-        this.topic = n.topic;
-        this.qos = parseInt(n.qos);
-        this.subscriptionIdentifier = n.subscriptionIdentifier;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901117
-        this.nl = n.nl;
-        this.rap = n.rap;
-        this.rh = n.rh;
+        /**@type {MQTTInNode}*/const node = this;
+        /**@type {string}*/node.broker = n.broker;
+        /**@type {MQTTBrokerNode}*/node.brokerConn = RED.nodes.getNode(node.broker);
 
+        node.dynamicSubs = {};
+        node.isDynamic = n.hasOwnProperty("inputs") && n.inputs == 1
+        node.inputs = n.inputs;
+        node.topic = n.topic;
+        node.qos = parseInt(n.qos);
+        node.subscriptionIdentifier = n.subscriptionIdentifier;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901117
+        node.nl = n.nl;
+        node.rap = n.rap;
+        node.rh = n.rh;
 
-        if (isNaN(this.qos) || this.qos < 0 || this.qos > 2) {
-            this.qos = 2;
+        const Actions = {
+            CONNECT: 'connect',
+            DISCONNECT: 'disconnect',
+            SUBSCRIBE: 'subscribe',
+            UNSUBSCRIBE: 'unsubscribe',
+            GETSUBS: 'getSubscriptions',
+        };
+        const allowableActions = Object.values(Actions);
+
+        if (isNaN(node.qos) || node.qos < 0 || node.qos > 2) {
+            node.qos = 2;
         }
-        this.broker = n.broker;
-        this.brokerConn = RED.nodes.getNode(this.broker);
-        if (!/^(#$|(\+|[^+#]*)(\/(\+|[^+#]*))*(\/(\+|#|[^+#]*))?$)/.test(this.topic)) {
-            return this.warn(RED._("mqtt.errors.invalid-topic"));
+        if (!node.isDynamic && !isValidSubscriptionTopic(node.topic)) {
+            return node.warn(RED._("mqtt.errors.invalid-topic"));
         }
-        this.datatype = n.datatype || "utf8";
-        var node = this;
-        if (this.brokerConn) {
-            let v5 = this.brokerConn.options && this.brokerConn.options.protocolVersion == 5;
-            this.status({fill:"red",shape:"ring",text:"node-red:common.status.disconnected"});
-            if (this.topic) {
-                node.brokerConn.register(this);
-                let options = { qos: this.qos };
-                if(v5) {
-                    // options.properties = {};
-                    // if(node.userProperties) {
-                    //     let userProperties = RED.util.evaluateNodeProperty(node.userProperties, node.userPropertiesType, node, {});
-                    //     setUserProperties(userProperties, options.properties);
-                    // }
-                    // setIntProp(node,options.properties,"subscriptionIdentifier", 1);
-                    setIntProp(node, options, "rh");
-                    if(node.nl === "true" || node.nl === true) options.nl = true;
-                    else if(node.nl === "false" || node.nl === false) options.nl = false;
-                    if(node.rap === "true" || node.rap === true) options.rap = true;
-                    else if(node.rap === "false" || node.rap === false) options.rap = false;
+        node.datatype = n.datatype || "utf8";
+        if (node.brokerConn) {
+            const v5 = node.brokerConn.options && node.brokerConn.options.protocolVersion == 5;
+            setStatusDisconnected(node);
+            if (node.topic || node.isDynamic) {
+                node.brokerConn.register(node);
+                if (!node.isDynamic) {
+                    let options = { qos: node.qos };
+                    if(v5) {
+                        setIntProp(node, options, "rh", 0, 2, 0);
+                        if(node.nl === "true" || node.nl === true) options.nl = true;
+                        else if(node.nl === "false" || node.nl === false) options.nl = false;
+                        if(node.rap === "true" || node.rap === true) options.rap = true;
+                        else if(node.rap === "false" || node.rap === false) options.rap = false;
+                    }
+
+                    node.brokerConn.subscribe(node.topic,options,function(topic, payload, packet) {
+                        subscriptionHandler(node, node.datatype, topic, payload, packet);
+                    },node.id);
                 }
-
-                this.brokerConn.subscribe(this.topic,options,function(topic,payload,packet) {
-                    // node.debug(`Sent ${topic}, datatype ${node.datatype} `+JSON.stringify(packet));//TODO: remove
-                    if (node.datatype === "buffer") {
-                        // payload = payload;
-                    } else if (node.datatype === "base64") {
-                        payload = payload.toString('base64');
-                    } else if (node.datatype === "utf8") {
-                        payload = payload.toString('utf8');
-                    } else if (node.datatype === "json") {
-                        if (isUtf8(payload)) {
-                            payload = payload.toString();
-                            try { payload = JSON.parse(payload); }
-                            catch(e) { node.error(RED._("mqtt.errors.invalid-json-parse"),{payload:payload, topic:topic, qos:packet.qos, retain:packet.retain}); return; }
-                        }
-                        else { node.error((RED._("mqtt.errors.invalid-json-string")),{payload:payload, topic:topic, qos:packet.qos, retain:packet.retain}); return; }
-                    } else {
-                        if (isUtf8(payload)) { payload = payload.toString(); }
-                    }
-                    var msg = {topic:topic, payload:payload, qos:packet.qos, retain:packet.retain};
-                    if(v5 && packet.properties) {
-                        //msg.properties = packet.properties;
-                        setStrProp(packet.properties, msg, "responseTopic");
-                        setBufferProp(packet.properties, msg, "correlationData");
-                        setStrProp(packet.properties, msg, "contentType");
-                        // setIntProp(packet.properties, msg, "topicAlias", 1, node.brokerConn.serverProperties.topicAliasMaximum || 0);
-                        // setIntProp(packet.properties, msg, "subscriptionIdentifier", 1, 268435455);
-                        setIntProp(packet.properties, msg, "messageExpiryInterval", 0);
-                        setBoolProp(packet.properties, msg, "payloadFormatIndicator");
-                        setStrProp(packet.properties, msg, "reasonString");
-                        setUserProperties(packet.properties.userProperties, msg);
-                    }
-                    if ((node.brokerConn.broker === "localhost")||(node.brokerConn.broker === "127.0.0.1")) {
-                        msg._topic = topic;
-                    }
-                    node.send(msg);
-                }, this.id);
-                if (this.brokerConn.connected) {
+                if (node.brokerConn.connected) {
                     node.status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
                 }
             }
             else {
-                this.error(RED._("mqtt.errors.not-defined"));
+                node.error(RED._("mqtt.errors.not-defined"));
             }
-            this.on('close', function(removed, done) {
-                if (node.brokerConn) {
-                    node.brokerConn.unsubscribe(node.topic,node.id, removed);
-                    node.brokerConn.deregister(node,done);
-                }
-            });
-        } else {
-            this.error(RED._("mqtt.errors.missing-config"));
-        }
-    }
-    RED.nodes.registerType("mqtt in",MQTTInNode);
+            node.on('input', function (msg, send, done) {
+                const v5 = node.brokerConn.options && node.brokerConn.options.protocolVersion == 5;
+                const action = msg.action;
 
-    function MQTTOutNode(n) {
-        RED.nodes.createNode(this,n);
-        this.topic = n.topic;
-        this.qos = n.qos || null;
-        this.retain = n.retain;
-        this.broker = n.broker;
-        this.responseTopic = n.respTopic;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901114
-        this.correlationData = n.correl;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901115
-        this.contentType = n.contentType;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901118
-        this.messageExpiryInterval = n.expiry; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901112
-        try {
-            if (/^ *{/.test(n.userProps)) {
-                //setup this.userProperties
-                setUserProperties(JSON.parse(n.userProps), this);//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901116
-            }
-        } catch(err) {}
-        // this.topicAlias = n.topicAlias; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901113
-        // this.payloadFormatIndicator = n.payloadFormatIndicator; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901111
-        // this.subscriptionIdentifier = n.subscriptionIdentifier;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901117
-
-        this.brokerConn = RED.nodes.getNode(this.broker);
-        var node = this;
-        var chk = /[\+#]/;
-
-        if (this.brokerConn) {
-            let v5 = this.brokerConn.options && this.brokerConn.options.protocolVersion == 5;
-            this.status({fill:"red",shape:"ring",text:"node-red:common.status.disconnected"});
-            this.on("input",function(msg,send,done) {
-                if (msg.qos) {
-                    msg.qos = parseInt(msg.qos);
-                    if ((msg.qos !== 0) && (msg.qos !== 1) && (msg.qos !== 2)) {
-                        msg.qos = null;
-                    }
-                }
-                msg.qos = Number(node.qos || msg.qos || 0);
-                msg.retain = node.retain || msg.retain || false;
-                msg.retain = ((msg.retain === true) || (msg.retain === "true")) || false;
-                /** If node property exists, override/set that to property in msg  */
-                let msgPropOverride = function(propName) { if(node[propName]) { msg[propName] = node[propName]; } }
-                msgPropOverride("topic");
-                if(v5) {
-                    if(node.userProperties) {
-                        msg.userProperties = node.userProperties;
-                    }
-                    if(node.responseTopic) {
-                        msg.responseTopic = node.responseTopic;
-                    }
-                    if(node.correlationData) {
-                        msg.correlationData = node.correlationData;
-                    }
-                    if(node.contentType) {
-                        msg.contentType = node.contentType;
-                    }
-                    if(node.messageExpiryInterval) {
-                        msg.messageExpiryInterval = node.messageExpiryInterval;
-                    }
-                    //Next, update/override the msg.xxxx properties from node config
-                    //TODO: Should we be expecting msg.properties.xxxx instead of msg.xxxx?
-                    // setStrProp(node,msg,"responseTopic");
-                    // setBufferProp(node,msg,"correlationData");
-                    // setStrProp(node,msg,"contentType");
-                    // setIntProp(node,msg,"messageExpiryInterval");
-                    //FUTURE setStrProp(node,msg,"topicAlias");
-                    //FUTURE setBoolProp(node,msg,"payloadFormatIndicator");
-                    //FUTURE setIntProp(node,msg,"subscriptionIdentifier");
-                }
-                if (msg.userProperties && typeof msg.userProperties !== "object") {
-                    delete msg.userProperties;
-                }
-                if (msg.hasOwnProperty("topicAlias") && !isNaN(msg.topicAlias) && (msg.topicAlias === 0 || node.brokerConn.serverProperties.topicAliasMaximum === 0 || msg.topicAlias > node.brokerConn.serverProperties.topicAliasMaximum)) {
-                    delete msg.topicAlias;
+                if (!allowableActions.includes(action)) {
+                    done(new Error(RED._('mqtt.errors.invalid-action-action')));
+                    return;
                 }
 
-                if ( msg.hasOwnProperty("payload")) {
-                    let topicOK = msg.hasOwnProperty("topic") && (typeof msg.topic === "string") && (msg.topic !== "");
-                    if (!topicOK && v5) {
-                        //NOTE: A value of 0 (in server props topicAliasMaximum) indicates that the Server does not accept any Topic Aliases on this connection
-                        if (msg.hasOwnProperty("topicAlias") && !isNaN(msg.topicAlias) && msg.topicAlias >= 0 && node.brokerConn.serverProperties.topicAliasMaximum && node.brokerConn.serverProperties.topicAliasMaximum >= msg.topicAlias) {
-                            topicOK = true;
-                            msg.topic = ""; //must be empty string
-                        } else if (msg.hasOwnProperty("responseTopic") && (typeof msg.responseTopic === "string") && (msg.responseTopic !== "")) {
-                            //TODO: if topic is empty but responseTopic has a string value, use that instead. Is this desirable?
-                            topicOK = true;
-                            msg.topic = msg.responseTopic;
-                            //TODO: delete msg.responseTopic - to prevent it being resent?
-                        }
-                    }
-                    if (topicOK) { // topic must exist
-                        // node.debug(`sending msg to ${msg.topic} `+JSON.stringify(msg));//TODO: remove
-                        this.brokerConn.publish(msg, function(err) {
-                            let args = arguments;
-                            let l = args.length;
-                            done(err);
-                        });  // send the message
+                if (action === Actions.CONNECT) {
+                    handleConnectAction(node, msg, done)
+                } else if (action === Actions.DISCONNECT) {
+                    handleDisconnectAction(node, done)
+                } else if (action === Actions.SUBSCRIBE || action === Actions.UNSUBSCRIBE) {
+                    const subscriptions = [];
+                    let actionData;
+                    //coerce msg.topic into an array of strings or objects (for later iteration)
+                    if(action === Actions.UNSUBSCRIBE && msg.topic === true) {
+                        actionData = Object.values(node.dynamicSubs);
+                    } else if (Array.isArray(msg.topic)) {
+                        actionData = msg.topic;
+                    } else if (typeof msg.topic == 'string' || typeof msg.topic == 'object') {
+                        actionData = [msg.topic];
                     } else {
-                        node.warn(RED._("mqtt.errors.invalid-topic"));
+                        done(new Error(RED._('mqtt.errors.invalid-action-badsubscription')));
+                        return;
+                    }
+                    //ensure each subscription is an object with topic etc
+                    for (let index = 0; index < actionData.length; index++) {
+                        let subscription = actionData[index];
+                        if (typeof subscription === 'string') {
+                            subscription = { topic: subscription };
+                        }
+                        if (!subscription.topic || !isValidSubscriptionTopic(subscription.topic)) {
+                            done(new Error(RED._('mqtt.errors.invalid-topic')));
+                            return;
+                        }
+                        subscriptions.push(subscription);
+                    }
+                    if (action === Actions.UNSUBSCRIBE) {
+                        subscriptions.forEach(function (sub) {
+                            node.brokerConn.unsubscribe(sub.topic, node.id);
+                            delete node.dynamicSubs[sub.topic];
+                        })
+                        //user can access current subscriptions through the complete node is so desired
+                        msg.subscriptions = Object.values(node.dynamicSubs);
+                        done();
+                    } else if (action === Actions.SUBSCRIBE) {
+                        subscriptions.forEach(function (sub) {
+                            //always unsubscribe before subscribe to prevent multiple subs to same topic
+                            if (node.dynamicSubs[sub.topic]) {
+                                node.brokerConn.unsubscribe(sub.topic, node.id);
+                                delete node.dynamicSubs[sub.topic];
+                            }
+
+                            //prepare options. Default qos 2 & rap flag true (same as 'mqtt in' node ui defaults when adding to editor)
+                            let options = {}
+                            setIntProp(sub, options, 'qos', 0, 2, 2);//default to qos 2 (same as 'mqtt in' default)
+                            sub.qos = options.qos;
+                            if (v5) {
+                                setIntProp(sub, options, 'rh', 0, 2, 0); //default rh to 0:send retained messages (same as 'mqtt in' default)
+                                sub.rh = options.rh;
+                                setBoolProp(sub, options, 'rap', true); //default rap to true:Keep retain flag of original publish (same as 'mqtt in' default)
+                                sub.rap = options.rap;
+                                if (sub.nl === 'true' || sub.nl === true) {
+                                    options.nl = true;
+                                    sub.nl = true;
+                                } else if (sub.nl === 'false' || sub.nl === false) {
+                                    options.nl = false;
+                                    sub.nl = false;
+                                } else {
+                                    delete sub.nl
+                                }
+                            }
+
+                            //subscribe to sub.topic & hook up subscriptionHandler
+                            node.brokerConn.subscribe(sub.topic, options, function (topic, payload, packet) {
+                                subscriptionHandler(node, sub.datatype, topic, payload, packet);
+                            }, node.id);
+                            node.dynamicSubs[sub.topic] = sub; //save for later unsubscription & 'list' action
+                        })
+                        //user can access current subscriptions through the complete node is so desired
+                        msg.subscriptions = Object.values(node.dynamicSubs);
                         done();
                     }
-                } else {
+                } else if (action === Actions.GETSUBS) {
+                    //send list of subscriptions in payload
+                    msg.topic = "subscriptions";
+                    msg.payload = Object.values(node.dynamicSubs);
+                    send(msg);
                     done();
                 }
             });
-            if (this.brokerConn.connected) {
+
+            node.on('close', function(removed, done) {
+                if (node.brokerConn) {
+                    if(node.isDynamic) {
+                        Object.keys(node.dynamicSubs).forEach(function (topic) {
+                            node.brokerConn.unsubscribe(topic, node.id, removed);
+                        });
+                        node.dynamicSubs = {};
+                    } else {
+                        node.brokerConn.unsubscribe(node.topic,node.id, removed);
+                    }
+                    node.brokerConn.deregister(node, done);
+                }
+            });
+        } else {
+            node.error(RED._("mqtt.errors.missing-config"));
+        }
+    }
+    RED.nodes.registerType("mqtt in",MQTTInNode);
+    //#endregion  "MQTTIn node"
+
+    //#region "MQTTOut node"
+    function MQTTOutNode(n) {
+        RED.nodes.createNode(this,n);
+        const node = this;
+        node.topic = n.topic;
+        node.qos = n.qos || null;
+        node.retain = n.retain;
+        node.broker = n.broker;
+        node.responseTopic = n.respTopic;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901114
+        node.correlationData = n.correl;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901115
+        node.contentType = n.contentType;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901118
+        node.messageExpiryInterval = n.expiry; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901112
+        try {
+            if (/^ *{/.test(n.userProps)) {
+                //setup this.userProperties
+                setUserProperties(JSON.parse(n.userProps), node);//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901116
+            }
+        } catch(err) {}
+        // node.topicAlias = n.topicAlias; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901113
+        // node.payloadFormatIndicator = n.payloadFormatIndicator; //https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901111
+        // node.subscriptionIdentifier = n.subscriptionIdentifier;//https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901117
+
+        /** @type {MQTTBrokerNode}*/node.brokerConn = RED.nodes.getNode(node.broker);
+
+        const Actions = {
+            CONNECT: 'connect',
+            DISCONNECT: 'disconnect',
+        };
+
+        if (node.brokerConn) {
+            setStatusDisconnected(node);
+            node.on("input",function(msg,send,done) {
+                if (msg.action) {
+                    if (msg.action === Actions.CONNECT) {
+                        handleConnectAction(node, msg, done)
+                    } else if (msg.action === Actions.DISCONNECT) {
+                        handleDisconnectAction(node, done)
+                    } else {
+                        done(new Error(RED._('mqtt.errors.invalid-action-action')));
+                        return;
+                    }
+                } else {
+                    doPublish(node, msg, done);
+                }
+
+            });
+            if (node.brokerConn.connected) {
                 node.status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
             }
             node.brokerConn.register(node);
-            this.on('close', function(done) {
+            node.on('close', function(done) {
                 node.brokerConn.deregister(node,done);
             });
         } else {
-            this.error(RED._("mqtt.errors.missing-config"));
+            node.error(RED._("mqtt.errors.missing-config"));
         }
     }
     RED.nodes.registerType("mqtt out",MQTTOutNode);
+    //#endregion "MQTTOut node"
 };
diff --git a/packages/node_modules/@node-red/nodes/locales/en-US/messages.json b/packages/node_modules/@node-red/nodes/locales/en-US/messages.json
index 21ab968d9..cda9c71eb 100755
--- a/packages/node_modules/@node-red/nodes/locales/en-US/messages.json
+++ b/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": {
diff --git a/packages/node_modules/@node-red/nodes/locales/en-US/network/10-mqtt.html b/packages/node_modules/@node-red/nodes/locales/en-US/network/10-mqtt.html
index 558635ec7..cb794e1dc 100644
--- a/packages/node_modules/@node-red/nodes/locales/en-US/network/10-mqtt.html
+++ b/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">