Fix new feature "mqtt keep subscription"

[Steve-Mcl]

Types of changes

• Bugfix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)

Proposed changes

Primarily, this PR is to make #4078 work. It may be that we decide to actually revert #4078 in which case this PR should be closed.

• fixes the current NR crash that occurs in beta-2 (see Changing settings in MQTT-IN node causes node-red to crash on next publish (3.1.0-beta.2) #4132)
• always call unsubscribe in node.on('close'
  • let the unsubscribe function manage actual (un)subscriptions
• refactor and resuse the same logic for creating a subscription into new function createSubscriptionObject
  • this reduces duplication and ensures same logic is used in subscribe and connect/reconnect
• track unsubscriptions in unsubscribeCandidates object and if the new "keep subscriptions" is active, check for changes to the SubscriptionObject. Any changes found means not only did the user deploy but that a subscription that was "kept" has also been modified - thus it is a valid unsub candidate.

Testing info

Demo Flow and instructions

[{"id":"53e9f7881f86fa73","type":"mqtt out","z":"0742b1daaf64f113","name":"","topic":"smac/test/retain","qos":"1","retain":"true","respTopic":"","contentType":"","userProps":"","correl":"","expiry":"","broker":"33d54b0c06d035ce","x":380,"y":280,"wires":[]},{"id":"a64960edfefabe0b","type":"inject","z":"0742b1daaf64f113","name":"disconnect","props":[{"p":"action","v":"disconnect","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":120,"y":280,"wires":[["53e9f7881f86fa73"]]},{"id":"0f3811101953044d","type":"inject","z":"0742b1daaf64f113","name":"connect","props":[{"p":"action","v":"connect","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":110,"y":320,"wires":[["53e9f7881f86fa73"]]},{"id":"bdfe7fedf9bd122d","type":"inject","z":"0742b1daaf64f113","name":"PUB: hi via conn1 ","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"hi via conn1 ","payloadType":"str","x":150,"y":240,"wires":[["53e9f7881f86fa73"]]},{"id":"25c5124a62f8acc9","type":"mqtt out","z":"0742b1daaf64f113","name":"","topic":"smac/test/retain","qos":"1","retain":"false","respTopic":"","contentType":"","userProps":"","correl":"","expiry":"","broker":"7489eee90363b43e","x":380,"y":420,"wires":[]},{"id":"7b65996461cd3e45","type":"inject","z":"0742b1daaf64f113","name":"disconnect","props":[{"p":"action","v":"disconnect","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":120,"y":420,"wires":[["25c5124a62f8acc9"]]},{"id":"f50ced36ef0a485d","type":"inject","z":"0742b1daaf64f113","name":"connect","props":[{"p":"action","v":"connect","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":110,"y":460,"wires":[["25c5124a62f8acc9"]]},{"id":"b825d72d0699ed44","type":"inject","z":"0742b1daaf64f113","name":"PUB: hi via conn2","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"\"hi via conn2. The time is now \" & $moment().format(\"hh:mm:ss.SSS\")","payloadType":"jsonata","x":150,"y":380,"wires":[["25c5124a62f8acc9"]]},{"id":"2e3e86784ac9360a","type":"mqtt in","z":"0742b1daaf64f113","name":"smac/test/#","topic":"smac/test/#","qos":"1","datatype":"auto-detect","broker":"33d54b0c06d035ce","nl":false,"rap":true,"rh":0,"inputs":0,"x":290,"y":520,"wires":[["f5b69a5caa10b5f3"]]},{"id":"f5b69a5caa10b5f3","type":"debug","z":"0742b1daaf64f113","name":"SUB smac/test/#","active":true,"tosidebar":true,"console":true,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":510,"y":520,"wires":[]},{"id":"fd76aa610c23692a","type":"comment","z":"0742b1daaf64f113","name":"conn1","info":"","x":550,"y":280,"wires":[]},{"id":"22336ecbfc4571a4","type":"comment","z":"0742b1daaf64f113","name":"conn2","info":"","x":550,"y":420,"wires":[]},{"id":"50fe9e22f97c8307","type":"comment","z":"0742b1daaf64f113","name":"1️⃣ connect to conn1 & conn2 \\n 2️⃣test publish conn1 - you should see 'hi from conn1' in SUB smac/test/# \\n 3️⃣ disconnect conn1 \\n 4️⃣ publish 3 messages to conn2\\n 5️⃣connect conn1 \\n 6️⃣ check 3 missed messages + 1 the original conn1 retained message in SUB smac/test/#","info":"","x":350,"y":120,"wires":[]},{"id":"33d54b0c06d035ce","type":"mqtt-broker","name":"smac-emqx-mqtt-v5-conn1","broker":"broker.emqx.io","port":"1883","clientid":"smac-emqx-mqtt-v5-conn1x","autoConnect":true,"usetls":false,"protocolVersion":"5","keepalive":"60","cleansession":false,"autoUnsubscribe":false,"birthTopic":"","birthQos":"0","birthRetain":"false","birthPayload":"","birthMsg":{},"closeTopic":"","closeQos":"0","closeRetain":"false","closePayload":"","closeMsg":{},"willTopic":"","willQos":"0","willRetain":"false","willPayload":"","willMsg":{},"userProps":"","sessionExpiry":"30"},{"id":"7489eee90363b43e","type":"mqtt-broker","name":"smac-emqx-mqtt-v5-conn2","broker":"broker.emqx.io","port":"1883","tls":"841140cc7ea16b1e","clientid":"smac-emqx-mqtt-v5-conn2","autoConnect":false,"usetls":false,"protocolVersion":"5","keepalive":"45","cleansession":false,"autoUnsubscribe":true,"birthTopic":"","birthQos":"0","birthRetain":"false","birthPayload":"","birthMsg":{},"closeTopic":"","closeQos":"0","closeRetain":"false","closePayload":"","closeMsg":{},"willTopic":"","willQos":"0","willRetain":"false","willPayload":"","willMsg":{},"userProps":"","sessionExpiry":""},{"id":"841140cc7ea16b1e","type":"tls-config","name":"","cert":"","key":"","ca":"","certname":"2cf8b83ef9022dc28f70fd5a52659d941c8fa8c8fa8e444cf44afd240ccfc8f0-certificate.pem.crt","keyname":"2cf8b83ef9022dc28f70fd5a52659d941c8fa8c8fa8e444cf44afd240ccfc8f0-private.pem.key","caname":"AmazonRootCA1.pem","servername":"","verifyservercert":false,"alpnprotocol":""}]

Checklist

• I have read the contribution guidelines
• For non-bugfix PRs, I have discussed this change on the forum/slack team.
• I have run grunt to verify the unit tests pass
• I have added suitable unit tests to cover the new/changed functionality

[knolleary]

It may be that we decide to actually revert #4078 in which case this PR should be closed.

Why so?

edit - I mean, why may we revert 4078? I haven't seen any discussion on that