# Is there a way to force MQTT nodes to re-subscribe from the flow?

**URL:** <https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051>\
**Category:** General\
**Created:** [12 March 2020 22:11 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051 "2020-03-12T22:11:29Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![techraf](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/techraf/32/17389_2.png) [@techraf](https://discourse.nodered.org/u/techraf)\
**Post date:** [12 March 2020 22:11 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/1 "2020-03-12T22:11:30Z")

</div>

Let's consider the following:

 ![image](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/b/f/bf002fb02993a24625718dc8a983185d59f66c6e.png)

```auto
[{"id":"f484ccd9.7480e","type":"mqtt in","z":"ced90757.b4e1e8","name":"","topic":"vDevice/State","qos":"2","datatype":"auto","broker":"d0a0bfc5.b4176","x":110,"y":220,"wires":[["7d4cd0fc.00e96"]]},{"id":"c14e8beb.ee34e8","type":"mqtt in","z":"ced90757.b4e1e8","name":"","topic":"vDevice/Telemetrics","qos":"2","datatype":"auto","broker":"d0a0bfc5.b4176","x":130,"y":280,"wires":[["3c727614.33e56a"]]},{"id":"3c727614.33e56a","type":"switch","z":"ced90757.b4e1e8","name":"","property":"retain","propertyType":"msg","rules":[{"t":"false"},{"t":"true"}],"checkall":"true","repair":false,"outputs":2,"x":330,"y":280,"wires":[["b19f805b.83a39"],["3c81bd84.3dfc12"]]},{"id":"3c81bd84.3dfc12","type":"change","z":"ced90757.b4e1e8","name":"","rules":[{"t":"set","p":"payload","pt":"msg","to":"\"(\" & msg.payload & \")\"","tot":"jsonata"}],"action":"","property":"","from":"","to":"","reg":false,"x":520,"y":300,"wires":[["b19f805b.83a39"]]},{"id":"7d4cd0fc.00e96","type":"switch","z":"ced90757.b4e1e8","name":"","property":"retain","propertyType":"msg","rules":[{"t":"false"},{"t":"true"}],"checkall":"true","repair":false,"outputs":2,"x":330,"y":220,"wires":[["753bff47.3299f"],["526cdae4.842964"]]},{"id":"526cdae4.842964","type":"change","z":"ced90757.b4e1e8","name":"","rules":[{"t":"set","p":"payload","pt":"msg","to":"\"(\" & msg.payload & \")\"","tot":"jsonata"}],"action":"","property":"","from":"","to":"","reg":false,"x":520,"y":240,"wires":[["753bff47.3299f"]]},{"id":"b24f40e8.e7898","type":"mqtt in","z":"ced90757.b4e1e8","name":"","topic":"vDevice/NetworkState","qos":"2","datatype":"auto","broker":"d0a0bfc5.b4176","x":140,"y":120,"wires":[["92b368e5.8e5238"]]},{"id":"92b368e5.8e5238","type":"switch","z":"ced90757.b4e1e8","name":"","property":"retain","propertyType":"msg","rules":[{"t":"false"},{"t":"true"}],"checkall":"true","repair":false,"outputs":2,"x":330,"y":120,"wires":[["e93e9089.cd42c"],["3f56ee40.1655e2"]]},{"id":"3f56ee40.1655e2","type":"change","z":"ced90757.b4e1e8","name":"","rules":[{"t":"set","p":"payload","pt":"msg","to":"\"(\" & msg.payload & \")\"","tot":"jsonata"}],"action":"","property":"","from":"","to":"","reg":false,"x":520,"y":140,"wires":[["e93e9089.cd42c"]]},{"id":"e93e9089.cd42c","type":"debug","z":"ced90757.b4e1e8","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","x":730,"y":120,"wires":[]},{"id":"753bff47.3299f","type":"debug","z":"ced90757.b4e1e8","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","x":730,"y":220,"wires":[]},{"id":"b19f805b.83a39","type":"debug","z":"ced90757.b4e1e8","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","x":730,"y":280,"wires":[]},{"id":"d0a0bfc5.b4176","type":"mqtt-broker","z":"","name":"localhost MQTT","broker":"localhosta","port":"1883","clientid":"","usetls":false,"compatmode":false,"keepalive":"60","cleansession":true,"birthTopic":"","birthQos":"0","birthPayload":"","closeTopic":"","closeQos":"0","closePayload":"","willTopic":"","willQos":"0","willPayload":""}]

```

Each flow reads a certain value from MQTT and passes it forward either:

- intact,
- or if the message had `msg.retain` set, it marks the payload as last-known-state (in the example: by enclosing it in parentheses).

* * *

Now, I would like to add logic that: if the device goes offline (the value of `vDevice/NetworkState` changes to `offline`), then the remaining values are "degraded" to last-known-state.

If I could force the MQTT-in nodes to re-subscribe, then they would get the values from MQTT with `msg.retain` flag set (automatically, by design) and pass the message according to existing logic.

Is it doable?

* * *

Or should I store the value of `vDevice/NetworkState` in context and modify each flow to consider messages matching (`retained: true` or `flow.networkstate: offline`) as last-known-state?

---

<div class="post-metadata">

**Author:** ![afelix](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/afelix/32/9743_2.png) [@afelix](https://discourse.nodered.org/u/afelix)\
**Post date:** [12 March 2020 22:15 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/2 "2020-03-12T22:15:07Z")

</div>

I’d say your last mentioned option is the best here, for exactly that reason. On top of that it’s also clear and easily maintainable, rather to somewhat obscure where you don’t know for sure whether at any point it has a subscription going. I’d say to keep the cat in the box where it is, but Schrödinger’s MQTT wouldn’t be my choice 😉  
Though maybe others think otherwise 🙂

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [13 March 2020 08:15 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/3 "2020-03-13T08:15:48Z")

</div>

When all is operating normally do the State and Telemetrics values come in from MQTT on a regular basis? If so then you can use a Trigger node to achieve what you want by making it send an appropriate message if no new data is received for a certain time.

Otherwise I would solve this by using a Join node in the mode described in the [cookbook example on joining streams](https://cookbook.nodered.org/basic/join-streams) to join the NetworkState in with the State, and the Telemetrics. Then you can access all the data you need in a Function node (or Change node) after the Join and send on the appropriate message each time a new message comes in.

---

<div class="post-metadata">

**Author:** ![techraf](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/techraf/32/17389_2.png) [@techraf](https://discourse.nodered.org/u/techraf)\
**Post date:** [13 March 2020 08:57 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/4 "2020-03-13T08:57:18Z")

</div>

> [@afelix](#):
>
> it’s also clear and easily maintainable

But I need to add a second message emitter into each node as change in the context value (to `offline`) wouldn't cause bottom flows to reevaluate.

Causing MQTT-in nodes to re-subscribe already deals with that.

On the other hand, it would be a cookie-cutter change to each flow, so not that bad.

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [13 March 2020 09:09 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/5 "2020-03-13T09:09:28Z")

</div>

> [@techraf](#):
>
> change in the context value (to `offline` ) wouldn't cause bottom flows to reevaluate.

Using a Join node instead of context solves that problem.

---

<div class="post-metadata">

**Author:** ![techraf](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/techraf/32/17389_2.png) [@techraf](https://discourse.nodered.org/u/techraf)\
**Post date:** [13 March 2020 09:09 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/6 "2020-03-13T09:09:58Z")

</div>

> [@Colin](#):
>
> When all is operating normally do the State and Telemetrics values come in from MQTT on a regular basis?

Yes, but with different periods for each flow. Worse yet, some values would come in with different periods over time. Say, if a device is active it sends values more often.

And unfortunately, `trigger` node "wait for"-value is set in its properties―there's no way to change it dynamically.

* * *

> [@Colin](#):
>
> Otherwise I would solve this by using a Join node

How?

`join` node needs to collect all values to pass them further, it doesn't remember the last state:

```auto
[{"id":"f7f35dce.1bb7f","type":"tab","label":"Flow 8","disabled":false,"info":""},{"id":"92864639.ab94f8","type":"join","z":"f7f35dce.1bb7f","name":"","mode":"custom","build":"object","property":"payload","propertyType":"msg","key":"topic","joiner":"\\n","joinerType":"str","accumulate":false,"timeout":"","count":"","reduceRight":false,"reduceExp":"","reduceInit":"","reduceInitType":"","reduceFixup":"","x":570,"y":240,"wires":[["8154828d.37646"]]},{"id":"a1914348.8f37a","type":"inject","z":"f7f35dce.1bb7f","name":"","topic":"value","payload":"1","payloadType":"num","repeat":"","crontab":"","once":false,"onceDelay":0.1,"x":150,"y":180,"wires":[["92864639.ab94f8"]]},{"id":"b98a1be9.d3d8f8","type":"inject","z":"f7f35dce.1bb7f","name":"","topic":"state","payload":"offline","payloadType":"str","repeat":"","crontab":"","once":false,"onceDelay":0.1,"x":170,"y":260,"wires":[["faaa41c0.e9de9"]]},{"id":"faaa41c0.e9de9","type":"change","z":"f7f35dce.1bb7f","name":"","rules":[{"t":"set","p":"complete","pt":"msg","to":"true","tot":"bool"}],"action":"","property":"","from":"","to":"","reg":false,"x":390,"y":280,"wires":[["92864639.ab94f8"]]},{"id":"8154828d.37646","type":"debug","z":"f7f35dce.1bb7f","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","x":750,"y":240,"wires":[]},{"id":"58516c1f.1599b4","type":"inject","z":"f7f35dce.1bb7f","name":"","topic":"state","payload":"online","payloadType":"str","repeat":"","crontab":"","once":false,"onceDelay":0.1,"x":170,"y":300,"wires":[["faaa41c0.e9de9"]]}]

```

 ![image](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/c/b/cbd22f539ca3a5084346d94c20dc130c0cc1a5a0.png)

Assuming the online/offline message sets `msg.complete`:

1. the device goes online (inject `online`)
2. the device sends a value (inject `1`) → a joined message is emitted
3. the device goes offline → a message is emitted, but the last-known-state value is lost

`join` node has no purpose in this scenario.

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [13 March 2020 09:17 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/7 "2020-03-13T09:17:06Z")

</div>

I haven't got any time at your flow at this moment, are you using the Join node in the mode described in the link? If so then every time it receives a message it sends on a message with the new information and the previous values of all the other inputs (as key/value pairs). So it absolutely does remember the previous values of all input streams.

**Edit** Just send the messages to it direct from mqtt and look at what you get out.

---

<div class="post-metadata">

**Author:** ![techraf](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/techraf/32/17389_2.png) [@techraf](https://discourse.nodered.org/u/techraf)\
**Post date:** [13 March 2020 09:55 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/9 "2020-03-13T09:55:16Z")

</div>

Thank you for insisting on your idea.

> [@Colin](#):
>
> every time it receives a message it sends on a message with the new information and the previous values of all the other inputs (as key/value pairs).

It does indeed do that with `and every subsequent message` option checked (which is not in the example on the linked page).

This functionality simplifies not only this, but a whole lot of other flows. Thank you!

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/1X/d073cd938eafa2e558d7c2cd59003b3ef4963033.png) [@system](https://discourse.nodered.org/u/system)\
**Post date:** [27 March 2020 09:55 UTC](https://discourse.nodered.org/t/is-there-a-way-to-force-mqtt-nodes-to-re-subscribe-from-the-flow/23051/10 "2020-03-27T09:55:26Z")

</div>

This topic was automatically closed 14 days after the last reply. New replies are no longer allowed.