Skip to content

mqtt://

MQTT 3.1.1 messaging via eclipse/paho.mqtt.golang. Pub/sub with QoS 0/1/2, retained messages, wildcards, Last Will and Testament, TLS, mTLS.

MQTT 5.0 is not supported here — the mqtt5:// scheme name is reserved for a future implementation. Use a broker that downgrades to 3.1.1 if you need both sides on the same wire.

URI shape

mqtt://<topic>?brokerUrl=tcp://host:1883&[options]

<topic> is the MQTT topic to publish to (producer) or subscribe to (consumer). MQTT wildcards (+ single-level, # multi-level) work for consumers; producers should publish on a concrete topic.

Parameters

Broker connection

Param Type Default Description
topic path — MQTT topic (path token).
brokerUrl string tcp://localhost:1883 Primary broker URL (tcp:// / ssl:// / ws:// / wss://).
url string — Alias for brokerUrl.
serverURIs string — Comma-separated failover broker URLs (reconnect rotation).
clientId string — MQTT client identifier. Empty = paho auto-generates.
connectionName string — Appended to clientId suffix for broker-management visibility.

Auth

Param Type Default Description
username string — CONNECT username.
userName string — Alias for username.
password string — CONNECT password.

TLS

Param Type Default Description
secure boolean false Force TLS handshake (also implied by ssl:///tls:///wss:// URLs).
tlsCAFile string — PEM CA bundle.
tlsCertFile string — PEM client cert (mTLS).
tlsKeyFile string — PEM client private key.
tlsInsecureSkipVerify boolean false Skip ALL server cert verification. Insecure.
tlsMinVersion / tlsMaxVersion enum — Pin TLS version (TLSv1.2 / TLSv1.3).
tlsCipherSuites string — Comma-separated Go cipher-suite names.
httpsHostnameVerificationEnabled boolean true Verify the certificate's hostname against the URL host. False maps to InsecureSkipVerify.

Session

Param Type Default Description
cleanSession boolean true Discard prior subscriptions/queued messages on connect (true) or resume (false).
keepAliveInterval duration 30s PINGREQ cadence. Bare integer = seconds.
connectionTimeout duration 30s Connect/subscribe/publish-token timeout.
maxReconnectDelay duration 10m Max delay between paho's auto-reconnect attempts.
automaticReconnect boolean true Enable paho's built-in auto-reconnect.
reconnect boolean true Helix-native alias; also gates the Helix-side reconnect loop on transport errors.
maxReconnectAttempts integer -1 Helix-side max attempts. -1 = unlimited.
reconnectTimeWait duration 2s Helix-side initial backoff.
maxInflight integer 0 Max in-flight QoS 1/2 messages while resuming. 0 = library default.
mqttVersion integer 4 3 = MQTT 3.1, 4 = MQTT 3.1.1. MQTT 5.0 not supported here.

Topic

Param Type Default Description
qos integer 1 QoS level (0/1/2) for publish + subscribe. Per-message override via mqtt.QOS header. Defaults to 1.
retained boolean false Producer: set Retained flag. Per-message override via mqtt.RETAINED header.

Last Will and Testament

Param Type Default Description
willTopic string — LWT topic. Empty = no LWT registered.
willPayload string — LWT payload (UTF-8 bytes).
willQos integer 0 LWT QoS.
willRetained boolean false LWT retained flag.

Headers

Param Type Description
headerIncludePattern string regex Pass only matching header names. Empty = all.
headerExcludePattern string regex Drop matching header names.

MQTT 3.x has no per-message wire headers — the filter applies to the Helix exchange headers injected by Helix (mqtt.TOPIC / QOS / RETAINED / ...).

Body shape

Producer

  • In body — published verbatim (BodyAsBytes()).
  • mqtt.QOS / mqtt.RETAINED headers override URI defaults per-message.

Consumer

For each received message:

  • In body — payload bytes.
  • In headers (subject to the include/exclude filter):
Header Description
mqtt.TOPIC The matched topic (after wildcard expansion).
mqtt.QOS Message QoS (0/1/2).
mqtt.RETAINED "true" if the message was retained.
mqtt.MESSAGE_ID MQTT packet ID.
mqtt.DUPLICATE "true" if the broker marked it a duplicate (QoS > 0 redelivery).

Examples

Subscribe to a sensor wildcard:

- id: telemetry
  from: mqtt://sensors/+/temperature?brokerUrl=tcp://broker:1883&qos=1&clientId=svc-telemetry
  steps:
    - to: postgres://readings/insert

Publish a retained message:

- setHeader:
    name: mqtt.RETAINED
    constant: "true"
- to: mqtt://status/online?brokerUrl=tcp://broker:1883&qos=1

Persistent session resumed across restart:

- from: mqtt://orders/#?brokerUrl=tcp://broker:1883&clientId=svc-orders&cleanSession=false&qos=2

TLS + mTLS:

- to: mqtt://prod/events?brokerUrl=ssl://mqtt.prod:8883&tlsCAFile=/etc/ssl/mqtt-ca.pem&tlsCertFile=/etc/ssl/client.crt&tlsKeyFile=/etc/ssl/client.key&clientId=svc-events

Last Will and Testament:

- to: mqtt://status?brokerUrl=tcp://broker:1883&clientId=svc-1&willTopic=status/svc-1/state&willPayload=offline&willRetained=true&willQos=1

Behavior notes

  • Default QoS=1. At-least-once delivery is the safer behaviour for most use cases.
  • cleanSession=false keeps the broker-side subscription state and queued messages across reconnects. Pair it with a stable clientId so the broker can correlate. On Stop() the consumer does not unsubscribe in this mode — that would defeat the persistent-session guarantee.
  • Auto-reconnect. Both paho's internal reconnect (automaticReconnect=true) and the Helix-level Reconnector are wired. The Helix layer guards against full broker outages with exponential backoff capped at reconnectTimeWait × N.
  • maxInflight=0 uses paho's default. Raise it for high-throughput QoS 1/2 publishers to avoid head-of-line blocking on resume.
  • No manual ack. paho doesn't expose ack/nack to user code (no manual-ack API). Errors during processing are observability-only; the broker considers the message delivered as soon as paho ACKs.
  • MQTT 5.0. Not supported here. The scheme name mqtt5:// is reserved for a future implementation that exposes properties, enhanced auth, and shared subscriptions.

Source

  • pkg/component/mqtt/mqtt.go — Component, Endpoint.
  • pkg/component/mqtt/{producer,consumer,will}.go — paho integration, LWT.
  • pkg/component/mqtt/{config,security,connection_pool,headers}.go — config, TLS+auth, pooling, header filter.

See also