MQTT v5 added first-class request/response support (Response Topic + Correlation Data properties), but today doing an MQTT request in Node-RED means hand-rolling it: publish on one topic, subscribe on another, generate and track correlation IDs, keep a map of in-flight requests keyed by correlation data, handle timeouts, and clean it all up. It works (I have done it several times), but it's a lot of function-node plumbing for something that should be simple!
Before / after (screenshot): the top flow is the DIY version - a lookup map keyed by correlation data, timeout monitoring, cleanup, and an mqttReply helper stashed in flow context. The bottom flow is the same thing with a single mqtt request node.
What it does
Publishes your msg as the request, then resolves with the correlated response as a single output message (or throws an errors on timeout).
Pure MQTT v5 request/response uses sets responseTopic + correlationData on the outbound publish and matches the reply by correlation data.
Reuses the existing mqtt-broker config node.
Behaviour / decisions I'd like feedback on
The MQTT Request nodes output message topic is swapped back so that the request topic stays in topic and original response topic is set to msg.requestTopic. This might sound normal, but the first time I implemented this, naturally, the response message arrives with its own msg.topic (set to the published response topic) - i felt this was odd and swapped them!
Response topic: The node recommends (and defaults to) a static topic string field since subscribe is done once and shared accross the broker. Optionally, it can be set via ENV VAR in the typedinput so that compound/${MY_VAR}/topics can be used. Additionally, it does support dynamic (msg.) subscribe per-request and tear down after the reply/timeout. Reasonable?
Correlation data: The default is set to auto-generate a UUID. Alternatively it can be provided in the message via a property - a tip warning is shown about collisions
QoS: defaults to 1 (not 0) since a dropped response would only surface as a timeout; 2 is not the default because it isn't universally supported (e.g. AWS IoT). Agree?
Message expiry: outbound request's messageExpiryInterval is tied to the timeout, so a stale request isn't acted on after we've given up. This is defaulted to a number 5 but can be set by ENV and can be sent in by a msg property
v5-only: the node hard-depends on v5. Currently: if the selected broker isn't v5 it disables itself, shows a warning status + edit-panel tip, and rejects input. Right call, or should it degrade some other way?
Notes
I will fight requests to change the topic to a plain text input (like current MQTT nodes)
Why? Because I love using the $ ENV typed input for things like {SITE_ID}/{DEPT_ID}/{SERVER_ID}/status
I deliberately excluded jsonata/flow/global on the topic, responseTopic & correlationData fields. It just adds complexity! For now, it is simple and we can always use a change node if the current options are not enough. I might consider adding extra options in a follow up iteration (or accept a PR for that)
Editor Screenshots:
Any objections, prior work I've missed on the forum / PR, or any edge cases (shared response topics across nodes, correlation-data collisions, broker quirks) I should think about before I throw a PR together (and invest time in unit tests)?
if ${} is present, it will substitute the corresponding environment variable into the result: For example, given the value "Hello ${FOO}" and the env var FOO is set to World, this results in the value "Hello World"
If memory serves, I've definitely done multiples! Dang brain not braining
It occurred to me during development then I completely forgot to look in to it. Any pointers to give me a head start (a code link, a variable name, even a current node that does this) @dceejay
Is the idea to add another core node or to do this with a separate mqttv5-request-response node package? Can't you create such a node package and leave it at that? Why and what is the reasoning to add a core node for this?
The fewer core nodes, the less the chance that something breaks in core and the less needs to be maintained in core.
Hey @jbudd I dont use Tasmota but I did look into this for you.
At this time, unless Tasmota upgrades to an MQTT v5 lib, not at this time. There is an open issue here but no one has suggested an alternative lib. A quick search reveals more than one candidate e.g. mqtt5nano, ESP32_MQTTv5, H4AsyncMQTT - maybe if you add these suggestions they will get traction?
Alternative solution - if the MQTT Request node were to be permitted on V3 brokers, we would have to provide a way to embed the necessary v5 properties in the payload itself e.g.
payload: {
"value": "your payload embeded in an object to allow other properties",
"_properties": {
"correlationData": "<correl>",
"responseTopic": "<restop>",
}
}
From there, you would setup a Berry Script. Berry allows you to subscribe to arbitrary MQTT topics, parse JSON payloads, execute local commands, and publish the results back dynamically.
v3 support could be a future iteration if there is enough interest - but not likely to land in first iteration
In a distributed system where data or operations live on separate edge devices, matching asynchronous responses back to the original triggering flow is incredibly clunky in MQTT v3.1.1.
A classic use case is on-demand commanding. Imagine a dashboard button or an API call that needs to query a specific device for a value, read a single database row from a remote server, or trigger a camera snapshot right now. Instead of subscribing to a continuous stream of data you don't need, you fire a single request and get the immediate answer back in line within the exact same flow. Couple that with multiple users performing operations at the some time and it gets complex very quickly
Here are some articles that do a better job of explaining than me:
What polling? This is 100% asynchronous and event-driven.
I see why you might think that if you imagine someone poorly implementing a loop to constantly poll a resource, but that isn't the purpose of an MQTT Request node. It standardises the request response pattern. Before v5, every engineering team invented their own workaround (like cramming IDs into JSON payloads). There was no protocol-level enforcement, leading to broken interoperability between different vendor systems.
Let me just finish with this: once you know you need this - you need it bad
Core node. It shares deep functionality with the existing MQTT core codebase, and I want it to be a first-class citizen.
Not cleanly. Doing this as an external package would require duplicating a large amount of the core code. It would lead to code divergence, maintenance fragmentation, and inevitably falling behind core updates.
To me, it doesn’t make sense to have native v5 MQTT support in Node-RED core while omitting one of the defining features of the v5 specification. Adding support for this pattern has been on my radar since I first introduced MQTT v5 to Node-RED core a few years ago. Recently, I’ve been integrating external systems where I had to handle this manually - it was cumbersome and clunky, which is exactly why a native solution is being proposed.
I totally appreciate that concern, and as a rule of thumb, it’s a valid point. However, in this case, an MQTT Request node conceptually belongs in core right alongside http request and tcp request, alongside its brother and sister mqtt in and mqtt out nodes.
FWIW there are no new code files being added/required in core (only modification to the network/mqtt* files). There are no new external dependencies or entirely new code subsystems being introduced. They use no additional runtime resources if you dont use them and minimal when you do. It is deliberately a separate node from the MQTT IN and MQTT OUT (so no breakage to existing nodes) and a clean dedicated design (for purpose) in both naming and feature (atomic).
Providing 1st class support for a foundational protocol pattern like this belongs in Node-RED core (IMHO)
... in over 20 years of using MQTT I have never knowingly needed it ... if I ever wanted something like that (before Node-RED) I would ensure the topic was a retained one, then just subscribe to the topic - data would be delivered - then unsubscribe - but again, that was incredibly rare.
If I really need to do single command response then things like http are "really useful" and well documented
But hey I can see it does complete a set of tools so by all means make it so.
Don't get me wrong, there are workarounds Dave (and I've used them!) - but having to build that custom logic on three different systems recently is what finally drove me to this.
The MQTT v3 approach of "retaining a topic, subscribing on the fly, and then unsubscribing" works fine for isolated or low-throughput tasks, but it does hit a wall - it gets incredibly tricky incredibly quickly when you layer on unknown/dynamic response topics, multitenant requirements, external system enforcements (to the v5 spec).
Probably you have considered this already, but while implementing this, bear in mind the operation if the broker and/or the response client are across a network. They may go offline temporarily and so messages can be queued at various points and may eventually get dropped or may be delivered some time later when the network recovers. This can lead, for example, to repeated responses from the same request, or a response to arrive long after the initial request has timed out.
A good example of feature-creep in MQTT. A message bus is, by design, asynchronous. Adding state (i.e. session) by coordinating request and response pairs is going beyond what a message bus should be doing - by design.
If I use a message bus, I don't expect it to be handling stateful request-response patterns. Strangely MQTT does do this. If I want request-response, then something like TCP/IP or HTTP would be more appropriate.
This is just a perspective of the MQTT specification, that it should be implemented if its available is clear. It's just unfortunate that such a feature made it into the specification.
Another use case is where you need to absolutely guarantee that every message from a client gets delivered to the other client, even if the network connection can fail temporarily. Currently this is complex to achieve, an MQTT response node would make it significantly simpler.
Ah ok, so MQTT handles the buffering of responses when networks drop away. Does the MQTT server than also handle the matching of requests/responses and send the client the appropriate pairing when it happens to have arrived?
I.e. so that the request-response pattern becomes transparent for the client.
It’s important to note that this data is not relevant to the MQTT broker but serves to identify the relationship between sender and receiver.
So the client is still responsible for matching the original request with the response using the extra correlationData field in MQTTv5. So the client has to maintain the state while the broker remains stateless.
So then the new mqtt node here needs to know how to match request/response pairs using correlationData - I assume this would be a "simple" look up in a lookup table somewhere.
Thanks Dave. That lead me straight to it - I updated code late last night but didnt get around to posting back that I added support for nodeMessageBufferMaxLength until now:
... if I ever wanted something like that (before Node-RED) I would ensure the topic was a retained one, then just subscribe to the topic - data would be delivered - then unsubscribe - but again, that was incredibly rare.
)
