feat: MQTT Request Node - official v5 request-response support#5860
Open
Steve-Mcl wants to merge 12 commits into
Open
feat: MQTT Request Node - official v5 request-response support#5860Steve-Mcl wants to merge 12 commits into
Steve-Mcl wants to merge 12 commits into
Conversation
Contributor
Author
Demo FlowScreenshots
|
6bcacb4 to
b786725
Compare
Contributor
|
I have done some fairly extensive testing, in particular the operation over network failure events. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Types of changes
Proposed changes
As discussed here: https://discourse.nodered.org/t/we-have-http-request-tcp-request-serial-request-any-objections-to-an-mqtt-request/101413
(UPDATE 2026-07-12) rebased this atop of the 5865-fix-tests-for-win branch (PR #5866) to pick up test fixes for windows
Overview
Adds a new core
mqtt requestnode that implements the MQTT v5request/response pattern (Response Topic + Correlation Data), plus the test
infrastructure to exercise the MQTT nodes without a live broker.
Major changes
This work was done in a specific order to avoid any breakage of existing MQTT nodes or flows:
In order of implementation:
Reworked the MQTT unit tests to run against a mock broker. Added an
in-process fake broker so the suite no longer requires a real broker, and
made the tests run by default. They still run against a real broker when
NR_MQTT_TESTS=1. Verified the full suite passes in both modes (not 100%true, 1 test was not working against real or fake - that was fixed in the process).
Added the
mqtt requestnode (editor UI, runtime, and built-in help).Confirmed the existing MQTT tests continue to pass against both the mock
and a real broker (the change touches shared code such as
subscriptionHandler).Added a test suite for the
mqtt requestnode covering config defaults,the v5 guard, request/response round-trips, correlation matching, timeouts,
in-flight collisions, QoS, and the buffer cap.
delivered:trueand verifies the late reply is ignored - runs against both the mock and a real broker) and a mock-only half-open test (times out withdelivered:falseand verifies the stuck outgoing packet is purged)(UPDATE 2026-07-12) Timeout behaviour is now end-to-end The timeout is measured from when the message is received and covers the whole exchange (subscribe, publish and waiting for the response), so the node always produces exactly one outcome - a response or an error - within the timeout, even if the broker connection is half-open. If no matching response arrives in time (or the request cannot be sent), the node reports an error. The original input message is passed to any catch node, with a
msg.mqttobject describing the outcome:msg.mqtt.timedOut(boolean) - alwaystruefor a timeoutmsg.mqtt.qos(number) - the QoS the request was published atmsg.mqtt.delivered(boolean) -trueif publishing was confirmed before the timeout. For QoS 1/2 this meansthe broker acknowledged receipt; for QoS 0 it only means the packet was written to the connection.
Use
msg.mqtt.deliveredto decide whether a retry is safe: whenfalse, the request was not confirmed sent (and,for QoS 1/2, is purged from the client's outgoing store so it will not be re-sent on reconnect).
Demo flows & Screenshots
I will add these in a follow up comment
Additional detail
The node
with it, shows a "requires MQTT v5" status, and rejects input with an error.
once at start-up and kept for the node's lifetime. A dynamic (
msg.) responsetopic is subscribed per-request and always torn down after the matching
response (or timeout).
uniqueness); can instead be taken from a
msgproperty.There was much discussion on the forum about hiding this altogether however
there are valid use cases for a user-specified correlation value (e.g. a
PLC with minimal string support, where the correlation is a simple integer representing
a sequence number, where correlation is a composite of multiple values, or where the
correlation data is exxpected to be a specific format).
The node supports both, and the default is a safe choice for most users.
surface as a timeout. QoS 2 is intentionally not the default as it isn't
universally supported (e.g. AWS IoT).
sent with a v5
messageExpiryIntervalequal to the wait timeout, so thebroker discards an undelivered request once we've stopped waiting for it.
(SUBACK) before the request is published; a refused subscription fails the
request instead of publishing into the void.
Message property pass-through
properties are preserved — including
req/res(http-in),_linkSource(link-call),
_msgid, and any user properties — so the node composes inlinein existing flows.
payload,topic,qos,retain,responseTopic,correlationData,contentType,messageExpiryInterval,payloadFormatIndicator,userProperties) are replaced with the response's,or cleared if the response doesn't include them (so a request-side value can't
linger and masquerade as the response's).
The response payload is always set in
msg.payload.RED.util.cloneMessage) up front, before anypublish-time mutation, and that clone is the base for the output — isolating
the in-flight message from by-reference changes elsewhere.
cloneMessagekeeps
req/resby reference, so http-response style flows still work.requestTopicmsg.topicon the output is the topic the response arrived on (the usualMQTT convention). To avoid losing the original request topic, it's preserved
as
msg.requestTopic.Error handling
error rather than emitting on its output. The clean, passed-through input
message (non-MQTT properties intact) is routed to any Catch node so the
failure can be handled.
Collisions and back-pressure
response topic + correlation data. Two requests that would put the same pair
on the same client are indistinguishable on the wire, so the second is
rejected rather than allowed to cross responses — this also covers multiple
request nodes sharing one broker.
nodeMessageBufferMaxLengthruntime setting.Test infrastructure
mqtt.connectis exported as a non-configurable getter and can't be stubbedwith
sinon, so the mock is installed by swapping themqttmodule's cachedexports for a shim whose
connectwe control; the node module is requiredafter the shim is in place.
responder (mqtt in → change → mqtt out) rather than test-internal hooks, so
the same tests pass against both the fake and a real broker.
correlation, and a forced SUBACK refusal) run only against the fake broker.
What that essentially means is that in day-to-day development, all tests are
ran (on the fake broker) but real tests against an actual MQTT broker (which
to be fair, were rarely run anyway) are still possible except for 2 tests
which would require a lot more plumbing for little gain.
Checklist
npm run testto verify the unit tests pass