natsmqtt5

README

natsmqtt5

An MQTT v5.0 broker that uses an existing NATS server as its messaging core.

nats-server speaks MQTT 3.1.1 and will not be adding v5: "we don't want to own the ongoing maintenance cost for the massive API surface increase that comes with MQTT 5 at this time." This is the other option raised in that thread — a separate broker that keeps NATS as the messaging engine, so you point it at the NATS cluster you already run rather than deploying a second, unrelated message bus.

Built against: MQTT 5.0 (OASIS, 2019-03-07 — still the current revision), nats.go v1.54.0 and nats-server v2.15.0, checked 2026-10-07. Both are the latest releases. Since v0.9.1 this module needs Go 1.26 (#11). Go 1.25 left upstream security support on 2026-08-19, and both NATS releases declare go 1.26.0. Holding the floor had kept golang.org/x/crypto below the fix for two x/crypto/ssh DoS advisories (GO-2026-6354, GO-2026-6355). See UPSTREAM.md, CHANGELOG.md for what each release changed under you, and CONFORMANCE.md for a statement-by-statement review against the MQTT 5 specification, and ROADMAP.md for where the broker knowingly does not yet conform, in the order it will be fixed.

MQTT v5 clients ──TCP/TLS──▶ natsmqtt5 ──▶ your existing NATS server
                             (a NATS client, not a fork)

What it does

• Speaks MQTT v5 to clients. All fifteen control packets, all 27 properties. Verified against the Eclipse Paho v5 client over a real socket.
• Uses NATS for everything underneath. Topics become NATS subjects, subscriptions become NATS subscriptions, shared subscriptions become NATS queue groups, retained messages live in a JetStream stream.
• Interoperates on the subject level. The topic-to-subject mapping is byte-for-byte the one nats-server uses for its own MQTT support, so sensors/7/temp lands on mqtt.sensors.7.temp either way. A NATS-native publisher reaches MQTT subscribers and vice versa, with no bridge.
• Scales horizontally by adding brokers. Two brokers against one NATS server are one logical broker: a client on either reaches subscribers on both.
• Keeps sessions across brokers, if you ask it to. With PersistentSessions, a client that reconnects with Clean Start 0 resumes its subscriptions on a broker that has never served it, and after a restart. Off by default; see Persistent sessions.
• Queues messages for a client that is away. QoS 1 and 2 messages published while a session is disconnected are delivered when it resumes, as MQTT 5 requires. On by default whenever JetStream is in use; see Offline queue.
• Speaks MQTT over WebSocket as well as TCP and TLS. Broker.WebSocketHandler() is an http.Handler; see MQTT over WebSocket.
• Embeds as a library. natsmqtt5.New returns a *Broker you run inside your own process, sharing your *nats.Conn if you want.

Install

As a container, against a NATS server you already run:

docker run --rm -p 1883:1883 \
  -e NATSMQTT5_NATS=nats://your-nats-server:4222 \
  ghcr.io/taumatix/natsmqtt5:v0.14.0

Images are published for linux/amd64 and linux/arm64 on every release, as :v0.14.0, :0.14.0, :0.14 and :latest. They are built from distroless/static, so there is no shell and no package manager in them, and the broker runs as a non-root user.

If you have no NATS server yet, compose.yaml starts one with JetStream enabled and the broker in front of it:

curl -O https://raw.githubusercontent.com/taumatix/natsmqtt5/v0.14.0/compose.yaml
docker compose up -d
mosquitto_pub -V 5 -h localhost -p 1883 -t sensors/7/temp -m 21.5

Every flag has an environment variable twin — upper-case, - becomes _, behind a NATSMQTT5_ prefix — so -subject-prefix is NATSMQTT5_SUBJECT_PREFIX. An explicit flag beats the environment. Run docker run --rm ghcr.io/taumatix/natsmqtt5:v0.14.0 -h for the full list.

As a binary:

go install github.com/taumatix/natsmqtt5/cmd/natsmqtt5@v0.14.0
natsmqtt5 -nats nats://localhost:4222 -listen :1883

As a library:

go get github.com/taumatix/natsmqtt5@v0.14.0

Requires Go 1.26 or newer (v0.8.0 and earlier build on Go 1.25), and a NATS server with JetStream enabled (or -no-retained / Options.DisableRetained to run without it).

Use

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

MQTT over WebSocket

Browsers and many hosted deployments reach MQTT only through WebSocket. Mount the broker's handler on any http.Server, plain or TLS; the connection joins the same broker, sessions and subscriptions as the TCP listener:

mux := http.NewServeMux()
mux.Handle("/mqtt", broker.WebSocketHandler())
go http.ListenAndServe(":8083", mux) // or ListenAndServeTLS

The handler negotiates the mqtt subprotocol and closes a client that offers none, and closes one that sends a text frame (MQTT-5.0 section 6). A browser's Origin must match the request's host or one of Options.WebSocketOriginPatterns. Broker.Close disconnects WebSocket clients with 0x8B like any other; shutting the http.Server down does not, because it cannot reach a hijacked connection. The TLS state and remote address an Authenticator sees are those of the HTTP request. A message is read up to the maximum packet size plus 64 KiB (for packets pipelined behind it); a larger one is refused with close status 1009 before it is buffered.

The command serves it with -ws-listen :8083 (path -ws-path, /mqtt by default). -tls-cert and -tls-key make both that listener (wss://) and -listen (mqtts://) serve TLS; there is no certificate reload, so restart the command to rotate.

Authentication and authorization are two small interfaces:

opts := natsmqtt5.Options{
	Authenticator: natsmqtt5.AuthenticatorFunc(
		func(ctx context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
			if !credentialsOK(req.Username, req.Password) {
				// Choose the CONNACK Reason Code the client sees.
				return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
			}
			// A Session is keyed by its Client Identifier alone, so an
			// authenticated user must not be free to choose any. Equality on
			// the first segment, not a prefix test: user "a" passes
			// HasPrefix("a/b/sensor", "a/") and would inherit "a/b"'s session.
			if owner, _, found := strings.Cut(req.ClientID, "/"); !found || owner != req.Username {
				return nil, &natsmqtt5.ConnectError{Code: packet.ClientIdentifierNotValid}
			}
			return &natsmqtt5.AuthResult{Identity: req.Username}, nil
		}),

	Authorizer: natsmqtt5.AuthorizerFunc(
		func(ctx context.Context, req *natsmqtt5.AuthzRequest) error {
			if req.Action == natsmqtt5.ActionPublish && !mayWrite(req.Identity, req.Topic) {
				return errors.New("denied") // becomes 0x87 Not authorized in the PUBACK
			}
			return nil
		}),
}

AuthRequest carries the TLS ConnectionState, so client-certificate authentication is a few lines.

The Client Identifier check above is not decoration. A Session is identified by its Client Identifier alone (MQTT-5.0 §4.1), so an Authenticator that verifies a principal but accepts any identifier lets that principal inherit another's session — its subscriptions, its unacknowledged messages and its QoS 2 receive state. Bind the identifier to the principal, or — for a client that sends a zero-length identifier to have one assigned [MQTT-3.1.3-6] — derive one with AuthResult.AssignClientID, as Example_authentication shows.

The broker re-runs the Authorizer over a resumed session's filters, with AuthzRequest.Resume set, so a permission narrowed between two connections takes effect on the second one rather than when the session expires. That bounds the damage; it does not remove the need to bind the identifier, and it has two costs worth knowing about. A slow Authorizer delays the CONNACK in proportion to the session's filter count. And a denial there is silent — there is no per-filter Reason Code in a CONNACK, so the filter is torn down and the client is still told SessionPresent: true.

A Will Message goes past the Authorizer too, at CONNECT, as an ActionPublish with AuthzRequest.Will set. The broker publishes a Will on the client's behalf after the client has gone, so CONNECT is the only moment it can be refused: a denied Will refuses the connection with 0x87, and the Will is not checked again when it fires.

A client that stays connected is not re-checked by itself, since asking on every delivery would put your Authorizer on the hot path of every message. When a principal's permissions change, call Broker.Reauthorize(ctx, clientID), from whatever tells you about revocations:

• each subscription goes back past the Authorizer (with Resume set), and a denied one stops delivering at once, including what was already queued;
• its unacknowledged messages are withdrawn, so a later resume does not resend them;
• the Will is checked again, and a denied one is discarded, even one already waiting out its Will Delay Interval.

The connection stays up, because a DISCONNECT would publish the Will being revoked. As at a resume, the client is not told which filters it lost, and any error from the Authorizer counts as a denial.

Any broker will do. Brokers sharing a NATS cluster and a SubjectPrefix answer one another on _NATSMQTT5.reauthorize.<prefix>, and the one holding the session runs the sweep. When none holds it, Reauthorize returns nil after ReauthorizeForwardWait (one second), since an offline session is re-checked when it next connects. Anything that can publish on that subject can make a broker re-ask your Authorizer, which can only narrow a session; restrict _NATSMQTT5.> to the brokers' NATS user if your other NATS clients are not trusted.

Offline queue

Core NATS delivers a message to whoever is subscribed when it is published and keeps nothing. MQTT 5 counts the QoS 1 and 2 messages pending for a session as part of its state, which outlives a disconnect while the Session Expiry Interval lasts [MQTT-3.1.2-23], so the broker keeps them in a JetStream stream and delivers them when the session resumes.

The queue is on by default whenever the broker uses JetStream: with retained messages (the default) or PersistentSessions. That default arrived in v0.10.0; up to v0.9.1 the queue is off unless OfflineQueue is set. A broker run with DisableRetained and no PersistentSessions does not touch JetStream and has no queue. If the queue's stream cannot be created, because the JetStream account is out of streams or storage for instance, the broker logs a warning and runs without it, retrying in the background (every 5 seconds, doubling to a minute) and logging at info when it comes up; Broker.OfflineQueueActive says whether it is on. The settings:

natsmqtt5.Options{OfflineQueueMaxAge: 24 * time.Hour} // on by default
natsmqtt5.Options{OfflineQueue: true}                 // required: fail to start without it
natsmqtt5.Options{DisableOfflineQueue: true}          // off

For the binary and the image they are -offline-queue-max-age, -offline-queue / NATSMQTT5_OFFLINE_QUEUE=true and -no-offline-queue / NATSMQTT5_NO_OFFLINE_QUEUE=true. Turning it off is a deviation from MQTT 5: a resumed session is told Session Present 1 and misses what was published while it was away.

The same stream serves a client that is connected but slower than its publishers. A connection holds up to 2048 messages waiting for its client; past that it reads the rest from the stream, at the client's pace and in order, and goes back to live delivery once it has caught up. Without the queue, QoS 1 and 2 messages past that point are dropped, as QoS 0 always is.

When the session resumes, the client gets what its filters matched while it was away, in publish order, before anything published after it reconnected. A message is delivered once: what the client received before it dropped and what arrives live during the resume are not repeated.

What it costs and what it does not cover yet:

• Each QoS 1 and 2 message is published twice: first under <prefix>.$queue.<subject> into the MQTT5_queue stream, which keeps it for OfflineQueueMaxAge (24 hours by default), and then live. A message whose live copy never reached the session (its publisher stopped between the two writes), or reached it after the session was released, is replayed however long ago it was stored: each session keeps a replay cursor, a queue sequence below which everything it was owed was sent except a short list of holes, and moves it by reading the stream without delivering once 1024 ids have piled up above it. That costs one read of the session's subjects per 1024 messages it is sent, and one JetStream request per SUBSCRIBE, to note where the new subscription starts (waited for at most 250 ms; without an answer the highest queue sequence the broker has seen is used, which can only owe the subscription more). A session without a cursor (one restored from a record written before it existed, or one that left more than 1024 holes or 16384 ids above it) replays from OfflineQueueRewind (10 seconds by default) before it was released instead, and starts a cursor once that replay has read the queue to its end. The broker waits for JetStream to confirm the first before publishing live and acknowledging the client, which adds a JetStream round trip to every QoS 1 and 2 publish. If JetStream cannot store it, the publish is refused, so the client tries again. QoS 0 is not queued.
• With PersistentSessions it also covers a session restored after a restart or on another broker: the stored record says when the session was released, and, for a client that was behind, the lowest queue sequence it had not been sent and the ids it had been sent above that (up to 4096 in the record, 65536 more beside it), so the replay starts where the client stopped, as for a session held in memory. Limits there. A record without that position (one from a connection that had delivered everything) replays from the release itself rather than a moment before it, because going back further could deliver a QoS 2 message twice; a message caught in the moments the connection went down can be lost that way. And a broker killed outright never releases its sessions, so a connected session writes its replay position to its record while it moves, once per Options.SessionCheckpointInterval (-session-checkpoint-interval, NATSMQTT5_SESSION_CHECKPOINT_INTERVAL; one second by default, negative turns it off). The next broker to resume the session sees it attached to an owner that no longer answers and replays from the last position written, so a kill costs the client the messages it was sent in that last interval, sent again, and loses none. A QoS 2 message is the exception: its record is written before the PUBLISH goes out and before the PUBREL does (about 0.3 ms each against a local NATS server, so a lockstep QoS 2 delivery takes about 0.77 ms instead of 0.16 ms; BenchmarkQoS2Delivery), so a kill at any step of the exchange resends the PUBLISH or the PUBREL and never delivers the message again as a new one [MQTT-4.3.3-6]. A QoS 1 message acknowledged in the last interval is sent again, as any at-least-once sender that dies before it records the PUBACK must. With the checkpoint off, or for a record written before it existed, the replay is everything the queue still holds, up to OfflineQueueMaxAge. The cost is one key-value write per interval for each client that received or was sent something in it.
• A message published straight onto NATS rather than through a broker is not queued (ROADMAP.md).
• A resume reads everything queued since the client left and matches it in the broker, so a long absence on a busy broker takes longer to catch up.

What the queue and a durable PUBACK cost

One client publishing QoS 1 (128-byte payload) and waiting for each PUBACK, over loopback TCP to an embedded NATS server with file-backed JetStream, on an Apple M2 Pro (go test -run '^$' -bench QoS1Publish -benchtime 3s, three runs each, see bench_publish_test.go):

	time per publish	publishes per second (one client)
no queue (DisableOfflineQueue)	about 34 µs	about 29,000
queue, the default	about 88 µs	about 11,000
queue and DurablePublish	about 111 µs	about 9,000

So the queue costs about 54 µs per QoS 1 publish here and DurablePublish about 23 µs more. Those are loopback numbers with a local disk: against a NATS server across a network the cost is the round trips, which is why the figures are not a forecast. The queue adds one JetStream round trip, DurablePublish one more, so add twice or three times your round trip to NATS, plus the server's fsync policy for file storage. A busy broker serves many clients at once, so total throughput is higher than one client's.

A PUBACK that means the message is safe

By default the broker waits for JetStream to store the queue's copy, then hands the live copy to its NATS connection and acknowledges at once. nats.go writes that from another goroutine, so a broker killed in that window loses a live copy it told the client it had. Options.DurablePublish (-durable-publish, NATSMQTT5_DURABLE_PUBLISH=true) closes it: for QoS 1 and 2 the PUBACK or PUBREC is sent only after (1) the queue stream stored the message (JetStream's publish acknowledgement) and (2) a flush of the broker's NATS connection came back, which means the NATS server has received the live publish. The flush says nothing about JetStream: the live copy is core NATS and is not stored by it, and the only stored copy is the queue's, from (1). If either fails or takes over 5 seconds the client gets 0x83 Implementation specific error and may send the message again. It implies the offline queue (the broker will not start without it) and is an error with DisableOfflineQueue. The default is off.

What the guarantee is: an acknowledged message is in the queue stream (for OfflineQueueMaxAge, to the stream's replication) and the NATS server has routed the live copy. A session that was away gets it from the queue; a connected subscriber gets it once, as without the option.

What it is not: live delivery stays core NATS, at most once per hop, so a connected subscriber whose own NATS path fails at the wrong moment can still miss it; ReconnectCatchUp below closes the case where that path is its broker's NATS connection (ROADMAP.md). A retry after a refusal can leave a second queue copy; Options.DurablePublishDedupe (-durable-publish-dedupe, NATSMQTT5_DURABLE_PUBLISH_DEDUPE=true, off by default) gives the publish a Nats-Msg-Id so the stream drops the retransmission (same packet identifier, DUP set, same topic and payload) inside its duplicate window, 2 minutes unless you changed the stream's. The live copy is still sent again. QoS 0 is not covered. Nor are Wills: a Will goes through the queue's keep like any other message but is not flushed, because there is no client to refuse to and nothing a refusal could ask it to retry; the record of a durable Will is already removed by the broker that publishes it (ROADMAP.md).

Messages published while a broker was off NATS

A broker whose NATS connection drops has no interest on the server until it reconnects, so a message another broker publishes meanwhile reaches none of its connected sessions, and by default nothing sends it later: only a session that was away is replayed. Options.ReconnectCatchUp (-reconnect-catch-up, NATSMQTT5_RECONNECT_CATCH_UP=true) closes that for QoS 1 and 2 messages that have a queue copy. When the connection drops, the broker holds its sessions' live QoS 1 and 2 copies back; when it is back, each connected session reads the queue from OfflineQueueRewind before the drop, or from just after the newest message the broker had seen if that is earlier (a server that died behind an open socket is noticed only when the pings fail). What the session has not been sent goes out in stream order, then live delivery resumes, so nothing newer overtakes it. The ids each session keeps stop a message going out twice, and a subscription is sent only what was stored after it was made. Off by default; an error with DisableOfflineQueue.

It costs one read of the queue stream per connected session after each reconnect, filtered to the session's subjects. It does not see a live copy lost between two servers of a NATS cluster while both brokers stay connected, nor messages published straight onto a live subject, which have no queue copy (publish them through NATSIngest instead; ROADMAP.md).

How MQTT maps onto NATS

MQTT	NATS
Topic sensors/7/temp	Subject mqtt.sensors.7.temp
Topic level /	Subject token separator .
A literal . in a topic	//
An empty topic level	./ or /. depending on position
Filter wildcard +	*
Filter wildcard #	>, plus a second subscription on the parent
Shared subscription $share/g/f	Durable JetStream consumer on the offline queue for QoS 1 and 2, pulled only by connected members; queue group on the subject for f for QoS 0
Retained message	JetStream stream, MaxMsgsPerSubject: 1
Session state (opt-in)	JetStream key-value bucket, one key per Client Identifier
v5 properties	NATS headers (Mqtt5-Content-Type, Mqtt5-User, …)

# gets two NATS subscriptions because MQTT's # matches the parent level (sport/# matches sport) and NATS's > does not.

The mqtt subject prefix is configurable with SubjectPrefix; brokers that should share traffic must agree on it, and brokers that must not share traffic must differ.

A NATS message with no Mqtt5-* headers is delivered as a QoS 0 message with no properties, which is what makes plain nats pub reach MQTT subscribers.

Publishing from plain NATS through a broker

A message published straight onto a live subject has no queue copy, whatever its Mqtt5-Qos header says: a session that was away never gets it, and a shared subscription gets it only through its NATS queue group, which can pick a member whose client is away and drop it there. With Options.NATSIngest (-nats-ingest, NATSMQTT5_NATS_INGEST=true) the brokers also listen on <prefix>.$ingest.<subject>, in one NATS queue group, and the broker that takes a message handles it as an MQTT PUBLISH on <subject>. It checks the topic and the QoS (and refuses a topic starting with $ whatever RestrictDollarTopics says, so the subject cannot write $retained or $queue), stores a retained message, stores the queue copy of a QoS 1 or 2 message, and publishes the live copy with its id and sequence. That copy reaches away sessions, shared groups and ReconnectCatchUp like any other.

nats request 'mqtt.$ingest.sensors.7.temp' 21.5 -H 'Mqtt5-Qos: 1'

The reply is empty, with Mqtt5-Msg-Id and, when queued, Mqtt5-Queue-Seq. If the broker refused the message, the reply carries Mqtt5-Ingest-Error with the reason, and nothing was published. A plain nats pub without a reply subject works too, but nothing tells it the outcome. Headers that do not start with Mqtt5- are passed on to the live copy. The Authorizer is not asked, because a NATS client has no MQTT identity: NATS permissions on <prefix>.$ingest.> decide who may publish there. Each broker handles ingest messages one at a time, so a publisher that waits for each reply keeps its order. Off by default, and then nothing answers on the subject.

Persistent sessions

By default a session lives in the broker process: a client reconnecting to the same broker within its Session Expiry Interval resumes its subscriptions, and one reconnecting to a different broker, or after a restart, has to subscribe again.

A session whose client never comes back is dropped once its Session Expiry Interval (capped by MaxSessionExpiry) has passed: a sweep, every SessionSweepInterval (default one minute), unsubscribes it from NATS and forgets it. A session can therefore outlive its interval by up to one sweep interval, never fall short of it, and a client that returns after the interval gets Session Present 0 whether or not the sweep has run.

Turn that off with one setting:

natsmqtt5.Options{PersistentSessions: true}

or -persistent-sessions / NATSMQTT5_PERSISTENT_SESSIONS=true for the binary and the image. Every broker that should share sessions needs it set, and needs the same StreamPrefix, since that names the bucket.

The subscription set is mirrored into a JetStream key-value bucket (MQTT5_sessions by default), keyed by Client Identifier. A CONNECT claims the record with a compare-and-swap on its revision, so of two brokers racing for one Client Identifier exactly one wins — and the loser disconnects its client with 0x8E Session taken over rather than serving it in parallel [MQTT-3.1.4-3]. That claim is also what makes the Session Present flag mean something across a restart.

compose.yaml runs two brokers this way:

NATSMQTT5_PERSISTENT_SESSIONS=true docker compose --profile cluster up -d

What it costs and what it does not cover:

• One JetStream write per CONNECT, SUBSCRIBE, UNSUBSCRIBE and DISCONNECT. None of those are the throughput path; PUBLISH is untouched.
• The Client Identifier is capped at 128 bytes (MaxPersistentClientIDLen), because it becomes part of a NATS subject. A longer one is refused with 0x85 Client Identifier not valid. Without persistence there is no limit.
• In-flight QoS 1 and QoS 2 messages are persisted as references. When a connection ends, the record keeps the Packet Identifier, QoS state and offline-queue sequence of each message the client has not acknowledged, and the identifiers of the client's QoS 2 PUBLISH packets received and not yet released. A session restored from the store, after a restart or on another broker, reads the payloads back from the queue and resends them with their original identifiers and DUP 1, resends the PUBREL for a QoS 2 exchange the client had taken ownership of, and acknowledges a resent QoS 2 PUBLISH without forwarding it twice. A message that expired in the meantime is dropped. Limits: a message with a copy in the offline queue is read back from it; one with none (retained, or with the queue off) travels in the record as its encoded PUBLISH when that is 16 KiB or less, and a larger one is written to the <StreamPrefix>_inflight bucket (named by the client and the SHA-256 of the PUBLISH, created on first use, deleted when the message is acknowledged, when the session ends or is replaced by a Clean Start, or when the expired record is swept, and aged out after twice the maximum Session Expiry Interval otherwise) and read back by the broker that restores the session. One whose queue copy has aged out, or whose PUBLISH exceeds the server's max_payload less 512 bytes, is logged and not resent; a broker killed outright leaves what its connections last checkpointed (-session-checkpoint-interval, one second): a QoS 1 message sent or acknowledged in that last interval is not (or no longer) in the record and is sent again, while a QoS 2 message is recorded before each packet that moves its exchange on; and a record is cut to fit the NATS server's max_payload (1 MiB by default), newest in-flight entries first, which is about 15000 entries, so a client with a Receive Maximum beyond that and that many messages unacknowledged loses the newest, logged.
• A filter the Authorizer no longer permits is dropped, and the CONNACK names it in a User Property that is this broker's own. Every filter of a resumed session goes back past the Authorizer before the CONNACK, on both branches, so a permission narrowed while the client was away takes effect immediately. A CONNACK has no per-filter Reason Code, so the CONNACK carries one natsmqtt5-dropped-filter User Property per dropped filter (natsmqtt5.DroppedFilterProperty); it is not part of MQTT v5, and a client that does not read it still sees SessionPresent: true and finds out by not receiving. It is left out when the CONNACK would exceed the client's Maximum Packet Size. The unacknowledged messages a dropped filter earned go with the filter.
• Clients may use $ topics unless you restrict them. MQTT 5.0 §4.7.2 says a server SHOULD prevent clients exchanging messages on topics starting with $. By default only $retained and $queue, where the broker keeps its own data, are refused, so applications using $app/... keep working. Set Options.RestrictDollarTopics, or -restrict-dollar-topics / NATSMQTT5_RESTRICT_DOLLAR_TOPICS=true, and a PUBLISH or Will on a $ topic is refused with Topic Name invalid (0x90) and a subscription to one with Topic Filter invalid (0x8F). $share/... stays allowed unless the filter inside it starts with $. A subscription a persistent session stored earlier is still resumed.
• The Will Message of a killed broker is published only with -durable-wills. Without it a Will belongs to the connection in memory, as before: a broker that shuts down cleanly publishes its clients' Wills, one killed outright does not. See Durable Will Messages.
• A session whose broker was killed outright is resumed whenever its client comes back, however long that takes. Expiry is measured from the moment a session is released, and a kill -9 records no release. A background sweep reclaims such a record once it is older than MaxSessionExpiry and its broker has stopped answering.

Durable Will Messages

A Will Message is published by the broker holding the connection, so a broker that is killed outright (kill -9, a lost machine) takes its clients' Wills with it. Set Options.DurableWills, or -durable-wills / NATSMQTT5_DURABLE_WILLS=true for the binary, and every connection's Will is also written to the <StreamPrefix>_wills key-value bucket, keyed by connection so a reconnect never overwrites its predecessor's record.

• A broker whose owner has stopped answering its liveness subject (two missed pings, so a busy broker is not mistaken for a dead one) has its Wills adopted by a survivor, which waits out the rest of the Will Delay Interval and publishes. Whoever wins a compare-and-swap delete of the record publishes it, so there is one publication however many brokers noticed [MQTT-3.1.2-8], [MQTT-3.1.2-10].
• A client that reconnects inside the delay cancels the Will if its session is resumed [MQTT-3.1.3-9]; one that starts a new session (Clean Start, or the session had ended) gets the old Will published first.
• A DISCONNECT with Reason Code 0x00 leaves nothing to publish [MQTT-3.1.2-8], [MQTT-3.1.2-10]; a Will on a live connection stays put.
• WillCheckInterval (-will-check-interval, default 5s) is how often a broker looks; the delay is counted from when a survivor notices, up to that long after the connection was lost.

The bucket has no time-to-live, because the record of a connection that stays open can be years old: a record is removed by the broker that publishes, cancels or drops it, or adopted by a survivor if that broker dies. A record that ended and has been overdue for more than 24 hours (counted from its due time, so no Will Delay Interval can outlast it) is deleted by any broker without being published, since what is left is one a fault leaked. Each broker's index holds a small fixed reference per record, not the Will, so many large Wills cost the bucket's storage and not every broker's memory.

It is off by default because the Will's payload sits in clear text in that bucket, and because it needs JetStream. Requirement: the Authorizer governs MQTT actions and cannot see the bucket, so whoever can read the key-value bucket <StreamPrefix>_wills (its $KV.<StreamPrefix>_wills.> subjects and $JS.API calls on it) can read every client's Will, and whoever can write it can have a Will published that no client set. Give the bucket the NATS permissions of the session bucket: the brokers' own NATS user and no other. It works with or without PersistentSessions. A broker cut off from NATS but still serving clients looks dead to the others, so it fences itself: if it cannot reach NATS (its connection, its own liveness subject and the Will bucket) for twice WillCheckInterval, it closes its client sockets, and a survivor adopts a silent owner's Wills only after the silence has outlasted that and the time the owner takes to notice (about 5 times WillCheckInterval, on top of the ping timeout). A Will is therefore published only after its client's connection is closed [MQTT-3.1.2-8]. The price is that the Will of a broker that really died waits that long longer than it did.

Deprecated: three defaults that flip in the next major version

Three behaviours are off by default today and will be on by default in the next major version. Nothing changes in v0.x; each can be pinned now, either way, with an option that beats the plain one. MIGRATION.md (a draft) says per flip what changes, who is affected and the one setting that keeps the old behaviour.

Behaviour	Pin it with	Binary
Durable Wills	Options.PinDurableWills	-durable-wills[=false]
Restricting $ topics	Options.PinRestrictDollarTopics	-restrict-dollar-topics[=false]
Outbound topic aliases	Options.PinOutboundTopicAliases	-outbound-topic-aliases[=false]

A broker with any of the three neither pinned nor on logs one INFO line per process naming them.

What it does not do

Stated plainly, because a broker you cannot trust the limits of is worse than one with fewer features:

• The offline queue needs JetStream, and covers what passes through a broker. A broker without JetStream, or with DisableOfflineQueue, drops QoS 1 and 2 messages for a session that is away, and messages published straight onto NATS are never queued. See Offline queue.
• A shared subscription hands QoS 1 and 2 work only to connected members, through a backlog in the offline queue. A QoS 1 message stays unacknowledged in the backlog until the member's PUBACK, and goes to another member if the member's session ends first. The backlog is deleted when no session is subscribed to the group any more (§4.8.2), tracked in a key-value bucket <StreamPrefix>_share_members. Without the queue it is a plain NATS queue group, which also chooses members whose client is away, and their share is dropped. Brokers sharing a SubjectPrefix must agree on the queue, or a message can reach a group twice. QoS 0 messages always go through the queue group.
• Without the queue, a client that falls behind loses QoS 1 and 2. A connection holds 2048 messages waiting for its client; with the queue on, a client slower than its publishers reads the rest from the stream, and with DisableOfflineQueue (or no JetStream) there is no copy to read, so the broker drops what does not fit and logs "dropping a message: the client is not keeping up". That is a deviation from [MQTT-4.1.0-1] and [MQTT-4.5.0-1] that the queue exists to close, and there is no option to hold publishers back instead: the only place to block is the NATS subscription's handler, which does not slow a publisher down, it moves the loss into the NATS client's pending buffer (and the server's slow-consumer handling of the broker's one connection, which every client shares). Run with the queue on if QoS 1 and 2 must not be lost to a slow client. Options.CloseBehindClients (-close-behind-clients, NATSMQTT5_CLOSE_BEHIND_CLIENTS=true, off by default) makes the loss visible instead: with no queue, the connection of a client that lost a QoS 1 or 2 message this way is closed with DISCONNECT 0x97 (Quota exceeded), and its session is kept to resume. The message is still lost, and so is anything published while the client is away; the client now knows.
• Retransmission on reconnect reaches as far as the broker's memory. A client that reconnects with Clean Start 0 to a broker still holding its session is resent the PUBLISH and PUBREL packets that connection left unacknowledged, with their original Packet Identifiers [MQTT-4.4.0-1]. A session restored from the store instead — after a restart, or on another broker — has nothing to resend.
• Shared subscriptions are granted QoS 1 by default. A QoS 2 request on a $share/ filter is answered with Granted QoS 1 in the SUBACK, which [MQTT-3.8.4-7] permits. Options.SharedQoS2 (-shared-qos2) grants QoS 2: the member that is sent a message owns it from then on, a held message survives a killed broker and a takeover, and the group is told at PUBREC. The guarantee is not complete (a message pulled but not yet checkpointed, or a broker killed between reading the PUBREC and writing the record, goes back to the group), so leave it off in production (ROADMAP.md). Options.SharedHoldLimit hands a message a connected member never acknowledges back to the group.
• Enhanced authentication is opt-in. With Options.Auther set, a CONNECT naming an Authentication Method runs the AUTH-packet exchange of MQTT-5.0 §4.12 (challenge, response, CONNACK) and a client may re-authenticate on the open connection (AUTH 0x19); a failed re-authentication ends in a DISCONNECT. A re-authentication that yields another identity re-checks the session's subscriptions and Will for it (as Reauthorize does) and drops what it no longer allows. hmacAuther in example_test.go is a complete reference method. Without an Auther a CONNECT carrying a method is refused with 0x8C Bad authentication method, as before. The Authenticator is not asked for such a CONNECT; the Authorizer sees the identity the exchange returns. The scram sub-package is a ready SCRAM-SHA-256 method (RFC 7677, no channel binding) over a credential lookup you supply: Options.Auther = &scram.Auther{Lookup: ..., OwnsClientID: ...}.
• The broker sends topic aliases only if asked to. It accepts them from clients always; Options.OutboundTopicAliases makes it send them to a client that offered a Topic Alias Maximum (off by default).
• Message Expiry Interval counts whole seconds. A message is deleted once it has waited its whole interval, and one still live goes out with the interval less the whole seconds it waited (2 s with 1.3 s waited leaves 1), on every delivery path including retained messages, which wait from the moment they were stored. The time is the broker's own clock and the JetStream timestamp, so brokers that share a NATS server agree to within clock skew between them. A retained message found expired is removed from the stream by the first subscription that looks; one nobody subscribes to stays until replaced. A live message's wait starts when the receiving broker got it, so time spent in NATS between brokers is free. Options.PublisherClockExpiry (-publisher-clock-expiry) counts it from the publishing broker's receipt instead, at the price of trusting the clocks: a publisher running behind, or a forged Mqtt5-Published-At header from anything that can publish on the subjects, makes a message expire early (a stamp in the future counts as no wait). Off by default.
• A broker's retained view can briefly serve a message another broker cleared. It is checked against the stream when the counts disagree and after every NATS reconnect it notices, but a partition shorter than the client's ping interval is neither. Options.RetainedVerifyInterval (-retained-verify-interval, off by default) also checks every entry on that timer, at one lookup per entry.
• A PUBACK does not mean the message reached NATS. The broker hands the message to its NATS connection and acknowledges; the write is flushed microseconds later. A broker killed in that window loses a QoS 1 message it has already acknowledged. A SUBACK, by contrast, does wait for the NATS server to confirm the subscription — otherwise a message published immediately after subscribing could be dropped, which is a loss the client has no way to notice.
• Topics containing a space, a tab, *, > or DEL are refused with 0x90 Topic Name invalid unless Options.EscapeNATSWildcards is set. MQTT permits them; a NATS subject cannot carry them. nats-server lets * and > through, which silently turns an MQTT subscription to a/*/b into a NATS wildcard that receives topics it must not see. Failing loudly is the default. Options.EscapeNATSWildcards (-escape-nats-wildcards / NATSMQTT5_ESCAPE_NATS_WILDCARDS=true) accepts them instead and writes each into the subject as / and a letter (* is /s, > is /g, space is /p, tab, line feed, carriage return, form feed and DEL are /t /n /r /f /d), a sequence the nats-server mapping never produces. Topics without those characters keep exactly the subject they had; the ones with them are no longer readable as the same subject by nats-server's own MQTT listener, so choose before you have data. Switching on needs nothing done: every topic the plain mapping could store has the same subject under the escaping one, so retained, queued and in-flight messages are found as before. Switching off has no rewrite to offer, because a topic that needed escaping cannot be written under the plain mapping at all, but what is stored is still read back under its real name (ret/a*b stays ret/a*b), so it keeps being served; clients can no longer publish to such a topic, nor subscribe to one by name, so publish an empty retained message to clear one before switching off if it should go.

Correctness

The MQTT layer is written against the OASIS MQTT v5.0 Standard of 07 March 2019, and the tests cite it:

• The codec tests assert the literal bytes from the spec's own worked examples (Figures 3-6, 3-9, 3-19, 3-21) and every boundary in Table 1-1, rather than only round-tripping our own encoder against our own decoder.
• The topic mapping is checked against all 40 conversion fixtures from nats-server's TestMQTTTopicAndSubjectConversion and TestMQTTFilterConversion, so a divergence from nats-server fails the build.
• The matching tests use the examples in §4.7.1.2, §4.7.1.3, §4.7.2 and §4.8.2.
• Integration tests run a real NATS server in-process with JetStream, a real broker, and the Eclipse Paho v5 client over a real TCP socket.
• The packet decoder and the topic mapping are fuzzed in CI.
• Every conformance statement a change touches is proved by an integration test against a real NATS server, citing its [MQTT-x.y.z-n] id; see CONTRIBUTING.md.

Packages

Package	What it is
github.com/taumatix/natsmqtt5	The broker
.../packet	MQTT v5 wire codec — standard library only, no NATS
.../topic	Topic validation, matching and the NATS subject mapping

packet and topic are usable on their own; neither imports NATS.

Licence

Apache 2.0. See LICENSE.

Documentation

Overview

Package natsmqtt5 is an MQTT version 5.0 broker that uses an existing NATS server as its messaging core.

The broker is a NATS client, not a fork of nats-server. Point it at a NATS URL you already run and it speaks MQTT v5 on a TCP port, translating topics to NATS subjects, fan-out to NATS subscriptions, shared subscriptions to NATS queue groups and retained messages to a JetStream stream. Two brokers against the same NATS server see each other's traffic, so an MQTT client on one reaches a subscriber on the other.

The subject mapping is byte-for-byte the one nats-server uses for its own MQTT 3.1.1 support, so topics land where a NATS-native subscriber expects.

Minimal use:

b, err := natsmqtt5.New(natsmqtt5.Options{NATSURL: "nats://localhost:4222"})
if err != nil {
	return err
}
defer b.Close()
return b.Serve(ctx)

Section numbers in the documentation refer to the OASIS MQTT v5.0 Standard of 07 March 2019.

Example

Run a broker against an existing NATS server.

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

Example (Authentication)

Reject unknown credentials with the CONNACK Reason Code of your choosing, and confine each user to its own topic subtree.

package main

import (
	"context"
	"errors"
	"log"
	"strings"

	"github.com/nats-io/nuid"

	"github.com/taumatix/natsmqtt5"
	"github.com/taumatix/natsmqtt5/packet"
)

func main() {
	passwords := map[string]string{"alice": "s3cret"}

	opts := natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",

		Authenticator: natsmqtt5.AuthenticatorFunc(
			func(_ context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
				if want, ok := passwords[req.Username]; !ok || want != string(req.Password) {
					return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
				}
				// A Session is keyed by its Client Identifier alone, so an
				// authenticated user must not be free to choose any: without
				// this, alice connecting as "bob-sensor" inherits bob's
				// session and everything it is subscribed to.
				//
				// A zero-length identifier is the client asking the server to
				// assign one [MQTT-3.1.3-6], so it is answered rather than
				// refused.
				if req.ClientID == "" {
					return &natsmqtt5.AuthResult{
						Identity:       req.Username,
						AssignClientID: req.Username + "/" + nuid.Next(),
					}, nil
				}
				// Equality on the first segment, not a prefix test: a user
				// named "a" passes HasPrefix("a/b/sensor", "a/") and would
				// inherit the session of the user named "a/b".
				if owner, _, found := strings.Cut(req.ClientID, "/"); !found || owner != req.Username {
					return nil, &natsmqtt5.ConnectError{Code: packet.ClientIdentifierNotValid}
				}
				return &natsmqtt5.AuthResult{Identity: req.Username}, nil
			}),

		Authorizer: natsmqtt5.AuthorizerFunc(
			func(_ context.Context, req *natsmqtt5.AuthzRequest) error {
				// A denied publish comes back as 0x87 in the PUBACK; a denied
				// subscription as 0x87 in the SUBACK. The connection stays up.
				if !strings.HasPrefix(req.Topic, "users/"+req.Identity+"/") {
					return errors.New("outside the user's subtree")
				}
				return nil
			}),
	}

	broker, err := natsmqtt5.New(opts)
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	log.Fatal(broker.Serve(context.Background()))
}

Example (Embedded)

Share one NATS connection between the broker and the rest of an application, and keep MQTT traffic in its own subject namespace.

package main

import (
	"context"
	"log"

	"github.com/nats-io/nats.go"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	// An existing connection the application already uses. The broker does not
	// close a connection it did not open.
	nc, err := nats.Connect("nats://localhost:4222")
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	broker, err := natsmqtt5.New(natsmqtt5.Options{
		Conn:          nc,
		Listen:        "127.0.0.1:1883",
		SubjectPrefix: "iot",
		StreamPrefix:  "IOT",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	// An MQTT client publishing to "sensors/7/temp" now reaches any NATS
	// subscriber on "iot.sensors.7.temp", and the other way round.
	log.Fatal(broker.Serve(context.Background()))
}

Example (EnhancedAuthentication)

Offer enhanced authentication: a CONNECT that names an Authentication Method is run through the Auther's AUTH-packet exchange, and the identity it ends with is the one the Authorizer sees.

package main

import (
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"errors"
	"log"
	"strings"

	"github.com/taumatix/natsmqtt5"
	"github.com/taumatix/natsmqtt5/packet"
)

// hmacAuther is a challenge/response method in the shape of SCRAM, small enough
// to read: the client names itself in its CONNECT's Authentication Data, the
// broker answers with a random nonce in an AUTH packet, and the client proves it
// holds the shared secret by returning HMAC-SHA256(secret, nonce). The secret
// never crosses the wire. It also serves the same exchange again for a
// re-authentication, with a fresh nonce.
type hmacAuther struct{ secrets map[string]string }

const hmacMethod = "HMAC-SHA256-CHALLENGE"

func (a hmacAuther) BeginAuth(_ context.Context, req *natsmqtt5.AuthBegin) (natsmqtt5.AuthExchange, error) {
	if req.Method != hmacMethod {
		return nil, natsmqtt5.ErrBadAuthMethod
	}
	return &hmacExchange{secrets: a.secrets, reauth: req.Reauth, user: req.Identity}, nil
}

type hmacExchange struct {
	secrets map[string]string
	reauth  bool
	user    string
	nonce   []byte
}

func (e *hmacExchange) Next(_ context.Context, data []byte) (natsmqtt5.AuthStep, error) {
	refused := &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
	if e.nonce == nil {

		if !e.reauth {
			e.user = string(data)
		}
		if _, ok := e.secrets[e.user]; !ok {
			return natsmqtt5.AuthStep{}, refused
		}
		e.nonce = make([]byte, 16)
		if _, err := rand.Read(e.nonce); err != nil {
			return natsmqtt5.AuthStep{}, err
		}
		return natsmqtt5.AuthStep{Data: e.nonce}, nil
	}
	mac := hmac.New(sha256.New, []byte(e.secrets[e.user]))
	mac.Write(e.nonce)
	if !hmac.Equal(mac.Sum(nil), data) {
		return natsmqtt5.AuthStep{}, refused
	}
	return natsmqtt5.AuthStep{Done: true, Identity: e.user, Data: []byte("ok")}, nil
}

func main() {
	opts := natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Auther:  hmacAuther{secrets: map[string]string{"alice": "s3cret"}},
		Authorizer: natsmqtt5.AuthorizerFunc(func(_ context.Context, req *natsmqtt5.AuthzRequest) error {
			if !strings.HasPrefix(req.Topic, req.Identity+"/") {
				return errors.New("not your subtree")
			}
			return nil
		}),
	}
	broker, err := natsmqtt5.New(opts)
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()
}

Index

• Constants
• Variables
• func Ptr[T any](v T) *T
• type Action
  • func (a Action) String() string
• type AuthBegin
• type AuthExchange
• type AuthRequest
• type AuthResult
• type AuthStep
• type Authenticator
• type AuthenticatorFunc
  • func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
• type Auther
• type Authorizer
• type AuthorizerFunc
  • func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error
• type AuthzRequest
• type Broker
  • func New(opts Options) (*Broker, error)
  • func NewWithContext(ctx context.Context, opts Options) (*Broker, error)
  • func (b *Broker) Close() error
  • func (b *Broker) ListenAddr() net.Addr
  • func (b *Broker) NATS() *nats.Conn
  • func (b *Broker) OfflineQueueActive() bool
  • func (b *Broker) Reauthorize(ctx context.Context, clientID string) error
  • func (b *Broker) Serve(ctx context.Context) error
  • func (b *Broker) WebSocketHandler() http.Handler
• type ConnectError
  • func (e *ConnectError) Error() string
• type Options

Examples

• Package
• Package (Authentication)
• Package (Embedded)
• Package (EnhancedAuthentication)

Constants

const (
	DefaultListen        = ":1883"
	DefaultSubjectPrefix = "mqtt"
	DefaultStreamPrefix  = "MQTT5"
	// DefaultReceiveMaximum bounds how many QoS 1 and QoS 2 publications a
	// client may have in flight towards the broker (MQTT-5.0 §3.2.2.3.3).
	DefaultReceiveMaximum = 1024
	// DefaultTopicAliasMaximum is the highest Topic Alias the broker accepts
	// from a client (MQTT-5.0 §3.1.2.11.5).
	DefaultTopicAliasMaximum = 64
	// DefaultMaximumPacketSize caps an inbound packet at 64 MiB
	// (MQTT-5.0 §3.2.2.3.6).
	DefaultMaximumPacketSize = 64 << 20
	// DefaultConnectTimeout bounds how long a new connection may stay silent
	// before sending its CONNECT (MQTT-5.0 §3.1.4, step 1).
	DefaultConnectTimeout = 10 * time.Second
	// DefaultMaxSessionExpiry caps the Session Expiry Interval a client may
	// request. The broker returns the capped value in CONNACK, as
	// MQTT-5.0 §3.2.2.3.2 provides for.
	DefaultMaxSessionExpiry = time.Hour
	// DefaultSessionCheckpointInterval is how often a broker with
	// PersistentSessions writes where a connected client stands to the session
	// record, when something has changed.
	DefaultSessionCheckpointInterval = time.Second
)

Defaults applied to the zero value of the corresponding Options field.

const DefaultRetainedTombstoneTTL = 24 * time.Hour

DefaultRetainedTombstoneTTL is how long a cleared retained topic's marker stays in the stream; see Options.RetainedTombstoneTTL.

const DefaultSessionSweepInterval = time.Minute

DefaultSessionSweepInterval is how often a broker looks for detached sessions whose Session Expiry Interval has passed; see Options.SessionSweepInterval.

const DroppedFilterProperty = "natsmqtt5-dropped-filter"

DroppedFilterProperty is the name of the User Property a CONNACK carries, once per filter, for each stored subscription a resumed session lost because the Authorizer refused it on resume. The CONNACK has no per-filter Reason Code, so a client that trusts Session Present would otherwise wait for messages that will not come. It is this broker's own and not part of MQTT v5; a client that does not read it loses nothing. It is left out when the CONNACK would not fit the client's Maximum Packet Size.

const MaxPersistentClientIDLen = 128

MaxPersistentClientIDLen caps the Client Identifier a broker with Options.PersistentSessions accepts. The identifier is encoded into a JetStream key-value key, and a key is part of a NATS subject, so it cannot be the 65535 bytes MQTT permits. A longer one is refused with 0x85 (Client Identifier not valid), which MQTT-5.0 §3.1.3.1 provides for: a server states the identifiers it accepts, and 128 bytes is more than five times the 23 that every MQTT server must accept.

const ReauthorizeForwardWait = time.Second

ReauthorizeForwardWait is how long Broker.Reauthorize waits for another broker to say it holds a session this one does not. Every broker answers at once, so the wait is spent only when no broker holds it.

Variables

var ErrAuthorizerUnavailable = errors.New("natsmqtt5: the Authorizer could not decide")

ErrAuthorizerUnavailable is what an Authorizer returns (bare or wrapped, it is matched with errors.Is) when it could not reach its policy and so cannot say either way. The broker then does not treat it as a denial where a denial would be wrong:

• a SUBSCRIBE gets Reason Code 0x83 in its SUBACK, a PUBLISH 0x83 in its PUBACK or PUBREC, and a CONNECT whose Will cannot be checked 0x83 in the CONNACK, instead of 0x87 Not authorized, so the client knows to retry;
• Broker.Reauthorize leaves the subscriptions and the Will it could not check as they are, and returns this error once the sweep is over;
• a CONNECT that resumes a session is refused with 0x83 before the session is claimed or the connection holding it displaced, so the stored filters and the live connection are untouched and the client can retry.

The resume check runs before the takeover, over the filters the session holds in memory and in the store. A filter that appears in between is checked again at resume, and an unavailable answer there keeps it: the filter was authorised when it was subscribed and no CONNECT is left to refuse.

var ErrBadAuthMethod = errors.New("natsmqtt5: unsupported authentication method")

ErrBadAuthMethod is what an Auther returns (bare or wrapped) for an Authentication Method it does not support: 0x8C Bad authentication method.

var ErrInvalidOptions = errors.New("natsmqtt5: invalid options")

ErrInvalidOptions reports an Options value the broker cannot run with.

Functions

func Ptr

func Ptr[T any](v T) *T

Ptr returns a pointer to v. The optional Options fields are pointers so that "not set" is distinguishable from a meaningful zero, and this saves callers declaring a variable to take the address of.

natsmqtt5.Options{MaximumQoS: natsmqtt5.Ptr(uint8(1))}

Types

type Action

type Action int

Action is an operation an Authorizer decides on.

const (
	// ActionPublish is a client publishing to a Topic Name.
	ActionPublish Action = iota
	// ActionSubscribe is a client subscribing to a Topic Filter.
	ActionSubscribe
)

func (Action) String

func (a Action) String() string

type AuthBegin

type AuthBegin struct {
	// Method is the Authentication Method the client named. Return
	// [ErrBadAuthMethod] from BeginAuth for one the Auther does not support.
	Method string
	// Data is the Authentication Data the client sent with it, which is the
	// first input of the exchange. It is also passed to the first Next call.
	Data []byte

	ClientID   string
	Username   string
	RemoteAddr net.Addr
	TLS        *tls.ConnectionState

	// Reauth is true for a re-authentication on an established connection
	// (§4.12.1); Identity is then what the connection authenticated as.
	Reauth   bool
	Identity string
}

AuthBegin describes the start of an enhanced authentication exchange.

type AuthExchange

type AuthExchange interface {
	// Next consumes the client's Authentication Data (the CONNECT's for the
	// first call, then each AUTH's) and says what happens next: a step with
	// Done false sends the step's Data to the client in an AUTH packet with
	// Reason Code 0x18 and waits for the client's reply; Done true accepts.
	//
	// A non-nil error rejects the exchange. A CONNECT is refused with the
	// [ConnectError]'s code (0x87 otherwise; [ErrBadAuthMethod] gives 0x8C); a
	// re-authentication ends in a DISCONNECT with that code, since the
	// connection can no longer be trusted [MQTT-4.12.1-2].
	Next(ctx context.Context, data []byte) (AuthStep, error)
}

AuthExchange is one run of an authentication method.

type AuthRequest

type AuthRequest struct {
	// ClientID is the Client Identifier from CONNECT, before the broker
	// assigns one for an empty value.
	ClientID string
	// Username is empty when the User Name Flag was not set.
	Username string
	// Password is nil when the Password Flag was not set. MQTT v5 permits a
	// password with no username (MQTT-5.0 §3.1.2.9).
	Password []byte
	// RemoteAddr is the client's network address.
	RemoteAddr net.Addr
	// TLS is the connection's TLS state, or nil for a plaintext connection.
	// Use it for client-certificate authentication.
	TLS *tls.ConnectionState
	// Properties are the CONNECT properties, including any User Properties.
	Properties *packet.Properties
}

AuthRequest describes a connection attempt for an Authenticator.

type AuthResult

type AuthResult struct {
	// Identity, when non-empty, names the authenticated principal. It is
	// passed to the Authorizer and included in log records; it does not change
	// the Client Identifier.
	Identity string
	// AssignClientID, when non-empty, overrides the Client Identifier for the
	// session and is returned to the client as the Assigned Client Identifier
	// [MQTT-3.1.3-7].
	AssignClientID string
}

AuthResult is an Authenticator's verdict.

type AuthStep

type AuthStep struct {
	// Data is sent as Authentication Data: in the AUTH that continues the
	// exchange, or, when Done, in the final CONNACK (or the AUTH 0x00 that ends a
	// re-authentication), where a method puts its server-final message.
	Data []byte
	// Done accepts the client.
	Done bool
	// Identity names the authenticated principal, as [AuthResult.Identity]. A
	// re-authentication must come out as the identity the connection already
	// has, or else the Authorizer is re-run over the session's subscriptions and
	// Will for the new identity, as Broker.Reauthorize does, and those it no
	// longer allows are dropped. If that check cannot be completed the client is
	// disconnected with 0x87.
	Identity string
	// AssignClientID is [AuthResult.AssignClientID]; CONNECT only.
	AssignClientID string
}

AuthStep is an AuthExchange's answer to the client's input.

type Authenticator

type Authenticator interface {
	Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
}

Authenticator decides whether a CONNECT is accepted.

Returning a nil error accepts the connection. Returning an error rejects it; wrap or return a *ConnectError to choose the CONNACK Reason Code, otherwise the broker answers 0x87 (Not authorized) and does not disclose the reason.

The Client Identifier is the session key

A Session is identified by its Client Identifier alone (MQTT-5.0 §4.1), so every connection an implementation accepts under a given identifier inherits that session: its subscriptions, its unacknowledged messages and its QoS 2 receive state. An implementation that authenticates a principal but lets it choose any ClientID therefore lets it read another principal's session.

So an implementation must either refuse a CONNECT whose AuthRequest.ClientID is not one the authenticated principal owns, or ignore it and return an AuthResult.AssignClientID derived from the principal. Nothing downstream can make up for this: the Authorizer decides topics, not sessions.

The broker re-runs the Authorizer over a resumed session's Topic Filters, so a permission narrowed between two connections takes effect on the second one rather than when the session expires. That bounds the damage; it does not replace binding the identifier, because the filters a principal is allowed are usually not what distinguishes it from the session's rightful owner.

type AuthenticatorFunc

type AuthenticatorFunc func(ctx context.Context, req *AuthRequest) (*AuthResult, error)

AuthenticatorFunc adapts a function to the Authenticator interface.

func (AuthenticatorFunc) Authenticate

func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)

Authenticate calls f.

type Auther

type Auther interface {
	BeginAuth(ctx context.Context, req *AuthBegin) (AuthExchange, error)
}

Auther drives enhanced authentication, the challenge/response exchange of MQTT-5.0 §4.12 that a CONNECT starts by naming an Authentication Method (SCRAM, Kerberos, a signed token) and that AUTH packets carry on. It is set in Options.Auther. With none set a CONNECT carrying an Authentication Method is refused with 0x8C (Bad authentication method), as before.

BeginAuth is called once per exchange, for the CONNECT (and again for every re-authentication a client starts, AuthBegin.Reauth). The exchange it returns holds that exchange's state; the broker calls it from the connection's own reader, so it is never used concurrently, and drops it when the exchange ends.

A CONNECT that names a method is decided by the Auther alone: the Authenticator, which sees only a user name and password, is not asked. The Authorizer is, with the identity the exchange returned.

As with an Authenticator, the Client Identifier names a session, and nothing downstream stops a principal from naming another's: BeginAuth must refuse a CONNECT whose AuthBegin.ClientID the authenticated principal does not own (or return an AuthStep.AssignClientID derived from it).

type Authorizer

type Authorizer interface {
	Authorize(ctx context.Context, req *AuthzRequest) error
}

Authorizer decides whether a session may publish to or subscribe to a topic. A nil error permits the operation.

A denied publish is answered with 0x87 (Not authorized) in the PUBACK or PUBREC, or dropped silently at QoS 0, as MQTT-5.0 §3.3.4 requires. A denied SUBSCRIBE is answered with 0x87 in the SUBACK for that filter. A denied Will Message, asked about with AuthzRequest.Will set, refuses the CONNECT with 0x87 in the CONNACK.

It is also called while a session is being resumed

A session is keyed by its Client Identifier alone, so the broker puts every filter of a resumed session back past the Authorizer before answering the CONNECT — once per live filter, with AuthzRequest.Resume set. That is what makes a permission narrowed between two connections take effect on the second one rather than when the session expires.

Two consequences an implementation has to plan for:

A slow Authorizer delays the CONNACK in proportion to how many filters the session holds. The ctx carries no deadline of the broker's making.

A denial there is silent and it is final. There is no per-filter Reason Code in a CONNACK, so the filter is torn down, logged at warn level, and the client is told SessionPresent: true with no indication that anything is missing. With Options.PersistentSessions the reduced set is written straight back to the durable record. An error returned because the policy store could not be reached is therefore indistinguishable from a decision to deny, and costs the client its subscription permanently — an implementation that cannot decide should return ErrAuthorizerUnavailable, which refuses the CONNECT with 0x83 and leaves the session as it was, rather than any other error.

type AuthorizerFunc

type AuthorizerFunc func(ctx context.Context, req *AuthzRequest) error

AuthorizerFunc adapts a function to the Authorizer interface.

func (AuthorizerFunc) Authorize

func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error

Authorize calls f.

type AuthzRequest

type AuthzRequest struct {
	Action Action
	// Resume distinguishes a filter being re-checked because a session is
	// being resumed from one the client has just asked for in a SUBSCRIBE. It
	// is only ever set with ActionSubscribe.
	//
	// The two are otherwise identical, and an implementation that audit-logs,
	// meters or rate-limits subscriptions needs to tell them apart: the client
	// performed no action, and the same filters come back on every reconnect.
	// It is also the cheap way out of the latency this check adds to a CONNACK
	// — an implementation may answer a resume from a cache it would not trust
	// for a fresh SUBSCRIBE.
	Resume bool
	// Will marks the Will Message of a CONNECT, asked about as an
	// ActionPublish before the connection is accepted. It is only ever set with
	// ActionPublish.
	//
	// The broker publishes a Will on the client's behalf after the client has
	// gone, so this is the only chance to refuse it: a denial refuses the
	// CONNECT with 0x87 (Not authorized), and the Will is not re-checked when it
	// fires. An implementation that audit-logs publishes can use the flag to
	// tell a Will — which may never be sent — from a PUBLISH the client made.
	Will bool
	// ClientID is the session's Client Identifier, after any assignment.
	ClientID string
	// Identity is the AuthResult.Identity from authentication, if any.
	Identity string
	// Username is the User Name from CONNECT, if any.
	Username string
	// Topic is the Topic Name for ActionPublish or the Topic Filter, shared
	// prefix included, for ActionSubscribe.
	Topic string
	// QoS is the requested QoS.
	QoS packet.QoS
	// Retain is set for a publish with the RETAIN flag.
	Retain bool
}

AuthzRequest describes an operation for an Authorizer.

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker is an MQTT v5 broker backed by a NATS server. Create one with New, run it with Serve, and release its resources with Close.

A Broker is safe for concurrent use.

func New

func New(opts Options) (*Broker, error)

New connects to NATS, provisions the JetStream assets the broker needs and binds the MQTT listener. It does not accept connections until Serve is called, so a caller can read ListenAddr first.

The returned Broker must be closed with Close.

func NewWithContext

func NewWithContext(ctx context.Context, opts Options) (*Broker, error)

NewWithContext is New with a context bounding the NATS connection and JetStream setup.

func (*Broker) Close

func (b *Broker) Close() error

Close stops the broker: it closes the listener, disconnects every client with 0x8B (Server shutting down) and releases the NATS resources it owns. A NATS connection passed in through Options.Conn is left open.

func (*Broker) ListenAddr

func (b *Broker) ListenAddr() net.Addr

ListenAddr reports the address the broker accepts MQTT connections on. It is valid as soon as New returns, which lets a test bind :0 and discover the port.

func (*Broker) NATS

func (b *Broker) NATS() *nats.Conn

NATS returns the NATS connection the broker uses, so an embedding application can share it rather than opening a second one.

func (*Broker) OfflineQueueActive

func (b *Broker) OfflineQueueActive() bool

OfflineQueueActive reports whether the broker is keeping QoS 1 and 2 messages for disconnected sessions. It is false when the queue is off by option, and when the queue is on only by default and could not be created at start-up (the broker keeps trying, and the value turns true once it succeeds).

func (*Broker) Reauthorize

func (b *Broker) Reauthorize(ctx context.Context, clientID string) error

Reauthorize re-runs the Authorizer over everything the session for clientID holds: each subscription, as at a resume, and its Will Message, as at CONNECT. Call it when a principal's permissions change; the broker does not poll a policy store it knows nothing about, and does not ask the Authorizer on every delivery.

A denied subscription is removed. It stops delivering at once, including messages already queued for the connection, and the messages it had in flight are withdrawn, so a later resume does not resend them; their acknowledgements are still accepted. A denied Will is discarded, including one already waiting out its Will Delay Interval. The connection itself is left alone: it is still allowed, and a DISCONNECT would publish the very Will being revoked. The client is not told which filters it lost; MQTT has no packet for that.

The Authorizer sees a subscription with AuthzRequest.Resume set and a Will with Will set, as it does for those checks elsewhere. As there, any error it returns is a denial, so an Authorizer that cannot reach its policy should not be asked to re-check a fleet during an outage.

Any broker will do. One that does not hold the session asks the others sharing its NATS connection and SubjectPrefix, and the one holding it runs the sweep and answers. When none holds it after ReauthorizeForwardWait (or when ctx ends first), it returns nil: a session held nowhere, only in the session store, is re-checked when it is next resumed. It returns ctx's error, having changed nothing further, if ctx ends during this broker's own sweep, and the holder's error if its sweep failed.

func (*Broker) Serve

func (b *Broker) Serve(ctx context.Context) error

Serve accepts MQTT connections until ctx is cancelled or Close is called. It returns nil on a clean shutdown.

func (*Broker) WebSocketHandler

func (b *Broker) WebSocketHandler() http.Handler

WebSocketHandler returns an http.Handler that serves MQTT over WebSocket (MQTT-5.0 §6). Mount it on any path of an http.Server (plain or TLS) and browsers and hosted clients can connect to the same broker, sessions and subscriptions as the TCP listener. Each upgraded connection is handed to the same state machine as a TCP one; nothing else differs.

The handler negotiates the "mqtt" subprotocol and refuses a client that offers none [MQTT-6.0.0-4]; every MQTT packet travels in a binary frame [MQTT-6.0.0-1]. A request whose Origin header names another host is refused unless Options.WebSocketOriginPatterns allows it.

The handler does not close the connections it serves when the http.Server shuts down (a hijacked connection is outside the server's reach); Close does, with the usual DISCONNECT 0x8B. A request made after Close is answered 503.

type ConnectError

type ConnectError struct {
	Code   packet.ReasonCode
	Reason string
}

ConnectError carries the CONNACK Reason Code an Authenticator wants the broker to send. Any refusal code MQTT-5.0 Table 3-1 lists for a CONNACK is sent as given; common choices are BadUserNameOrPassword (0x86), NotAuthorized (0x87), Banned (0x8A) and ClientIdentifierNotValid (0x85). A code the table does not list, success codes included, is logged and sent as UnspecifiedError (0x80) [MQTT-3.2.2-8].

func (*ConnectError) Error

func (e *ConnectError) Error() string

type Options

type Options struct {
	// NATSURL is the NATS server to use. Defaults to nats.DefaultURL. Ignored
	// when Conn is set.
	NATSURL string
	// NATSOptions are passed to nats.Connect for credentials, TLS and so on.
	// Ignored when Conn is set.
	NATSOptions []nats.Option
	// Conn is an already-established NATS connection to use instead of
	// dialling. The broker does not close a connection it did not open.
	Conn *nats.Conn

	// Listen is the TCP address to accept MQTT connections on. Defaults to
	// DefaultListen. Ignored when Listener is set.
	Listen string
	// Listener is an already-open listener to accept on instead of binding
	// Listen. The broker closes it on shutdown.
	Listener net.Listener
	// TLSConfig, when non-nil, wraps the listener in TLS.
	TLSConfig *tls.Config
	// WebSocketOriginPatterns lists the Origin hosts (path.Match patterns,
	// e.g. "app.example.com" or "*.example.com") a browser may connect from
	// through Broker.WebSocketHandler, besides the request's own host. A
	// request without an Origin header (any non-browser client) is always
	// accepted. Use "*" only if the broker's own authentication is what
	// protects it.
	WebSocketOriginPatterns []string

	// SubjectPrefix namespaces every NATS subject the broker publishes MQTT
	// traffic on, keeping it clear of NATS system subjects. Defaults to
	// DefaultSubjectPrefix. Set it to the same value on every broker that
	// should share traffic.
	SubjectPrefix string
	// StreamPrefix names the JetStream assets the broker creates. Defaults to
	// DefaultStreamPrefix. Brokers sharing a SubjectPrefix must share this.
	StreamPrefix string
	// DisableRetained turns off retained-message support, so no JetStream
	// assets are created and CONNACK advertises Retain Available = 0
	// (MQTT-5.0 §3.2.2.3.5). Use it against a NATS server without JetStream.
	DisableRetained bool
	// RetainedStorage selects file or memory storage for the retained-message
	// stream. Defaults to file storage.
	RetainedStorage jetstream.StorageType
	// RetainedReplicas is the JetStream replica count for the retained-message
	// stream. Defaults to 1.
	RetainedReplicas int
	// RetainedTombstoneTTL is how long the marker left in the retained stream
	// by clearing a retained message (an empty-payload retained PUBLISH) is
	// kept before the periodic sweep deletes it. Without the purge a workload
	// that retains and clears many distinct topics grows the stream without
	// bound. Defaults to DefaultRetainedTombstoneTTL; a negative value keeps
	// tombstones forever. A broker cut off from NATS for longer than this can
	// miss a clear and keep the cleared message until it restarts.
	RetainedTombstoneTTL time.Duration
	// RetainedVerifyInterval makes the periodic sweep check every entry of the retained
	// view against the stream at least this often (one lookup per entry), on top of the
	// count comparison it always makes and the full check after a NATS reconnect. It finds
	// a stale entry that a partition too short for the NATS client to notice left behind,
	// which neither of those can see [MQTT-3.3.1-6]. Zero, the default, turns it off. The
	// sweep runs every SessionSweepInterval, so the check is made at the first sweep after
	// the interval has passed.
	RetainedVerifyInterval time.Duration
	// SharedHoldLimit bounds how long a shared-subscription member may hold a
	// QoS 1 message without acknowledging it. Past it the message is handed
	// back to the group, so a member that is connected but never sends PUBACK
	// does not keep it from the others, and the group may deliver it twice:
	// the late PUBACK is still accepted and completes the exchange with that
	// client. Zero, the default, holds for as long as the session lives. The
	// check runs every third of the group's acknowledgement wait (10 s), so the
	// limit is met to within that.
	SharedHoldLimit time.Duration

	// PersistentSessions keeps session state in a JetStream key-value bucket
	// instead of in the broker process, so a client reconnecting with Clean
	// Start 0 resumes its subscriptions after a broker restart and on a broker
	// that has never seen it. Brokers sharing a StreamPrefix share the bucket
	// and arbitrate ownership between themselves, so exactly one of them serves
	// a given Client Identifier [MQTT-3.1.4-3].
	//
	// It needs JetStream, and it caps the Client Identifier at
	// MaxPersistentClientIDLen bytes.
	//
	// A resumed session carries its subscription set, and the offline queue
	// delivers what was published while it was away. Messages that were in
	// flight to the client when its broker went down are not stored, so they
	// are not resent. The Will Message is not part of the session record, because
	// it belongs to the network connection rather than to the session
	// (MQTT-5.0 §3.1.2.5); DurableWills stores it separately. ROADMAP.md has
	// the first.
	PersistentSessions bool
	// SessionStorage selects file or memory storage for the session bucket.
	// Defaults to file storage. Memory storage makes sessions survive a broker
	// restart but not a NATS restart.
	SessionStorage jetstream.StorageType
	// SessionReplicas is the JetStream replica count for the session bucket.
	// Defaults to 1.
	SessionReplicas int

	// RestrictDollarTopics follows MQTT-5.0 §4.7.2's "The Server SHOULD
	// prevent Clients from using such Topic Names to exchange messages with
	// other Clients" for every Topic Name and Topic Filter that starts with
	// "$": a PUBLISH or Will to one is refused with Topic Name invalid (0x90),
	// and a subscription to one with Topic Filter invalid (0x8F). A Shared
	// Subscription ($share/…) is allowed unless the Topic Filter inside it
	// starts with "$". Off by default, which leaves "$app/…" and the like usable
	// as before; "$retained" and "$queue", where the broker keeps its own data,
	// are refused either way. It does not apply to a subscription a persistent
	// session already stored.
	//
	// Deprecation notice: this is off by default today, and the default will
	// become on in the next major version (MIGRATION.md). Set
	// PinRestrictDollarTopics to choose now; leaving both unset logs one INFO
	// line per process.
	RestrictDollarTopics bool
	// PinRestrictDollarTopics fixes RestrictDollarTopics whatever the default
	// is, and beats RestrictDollarTopics when both are set: Ptr(false) keeps
	// "$app/…" usable after the default flips, Ptr(true) refuses it today.
	// nil leaves the choice to RestrictDollarTopics and the default.
	PinRestrictDollarTopics *bool

	// DurableWills keeps the Will Message of every open connection in a
	// JetStream key-value bucket, so that the Will of a broker that is killed
	// outright is still published: a surviving broker notices the owner has
	// stopped answering its liveness subject, adopts the Will, waits out
	// whatever is left of the Will Delay Interval and publishes it
	// [MQTT-3.1.2-7], [MQTT-3.1.2-8], [MQTT-3.1.3-9]. Exactly one broker publishes a given Will,
	// by compare-and-swap on the record.
	//
	// Off by default, because it changes where Wills live: the Will's payload
	// is written to the "<StreamPrefix>_wills" bucket in clear text, so that
	// bucket needs the access control of the session bucket. It needs JetStream
	// and works with or without PersistentSessions; without them a client that
	// reconnects to a different broker starts a new session, so a Will waiting
	// out its delay is published when it does, rather than cancelled.
	//
	// The delay is counted from the moment a surviving broker notices, which is
	// up to WillCheckInterval after the connection was lost.
	//
	// Deprecation notice: this is off by default today, and the default will
	// become on in the next major version (MIGRATION.md). Set PinDurableWills
	// to choose now; leaving both unset logs one INFO line per process.
	DurableWills bool
	// PinDurableWills fixes DurableWills whatever the default is, and beats
	// DurableWills when both are set: Ptr(false) keeps Wills in memory after
	// the default flips, Ptr(true) stores them durably today. nil leaves the
	// choice to DurableWills and the default.
	PinDurableWills *bool
	// WillCheckInterval is how often a broker looks for Wills whose owner has
	// gone, with DurableWills. It bounds how long after a broker's death its
	// clients' Wills wait to be noticed. Defaults to 5 seconds.
	//
	// It also sets the patience of a broker that loses NATS: one that cannot
	// have a write to the Will bucket accepted for twice this interval closes
	// its client connections, and a survivor waits about 5 intervals (plus the
	// liveness ping timeout, or at least two seconds below a one-second
	// interval) before adopting a silent broker's Wills, so a Will
	// is never published while its client's connection is open. A write needs a
	// majority of the bucket's servers, so a broker whose NATS server is cut off
	// from the rest of its cluster fences itself even though that server still
	// answers it. On a cluster, keep twice this interval above the time NATS
	// takes to elect a new leader for the bucket (it starts an election 4 to 9
	// seconds after the leader goes quiet): a shorter wait makes every broker
	// close its clients whenever the bucket's leader is lost.
	WillCheckInterval time.Duration

	// OfflineQueue requires the offline queue, so a broker that cannot set it
	// up fails to start.
	//
	// The queue keeps QoS 1 and QoS 2 messages for a session while its client
	// is disconnected, and delivers them when the session resumes, in publish
	// order and before anything published after the resume. MQTT-5.0 counts
	// those messages as session state that MUST outlive the connection
	// [MQTT-3.1.2-23], [MQTT-4.5.0-1], so the queue is on by default whenever
	// the broker uses JetStream (retained messages, the default, or
	// PersistentSessions); then a stream that cannot be created is logged at
	// warn level and the broker runs without it. Without the queue, messages
	// are dropped for an absent session, as core NATS keeps nothing.
	//
	// Each such message is published a second time, under
	// <SubjectPrefix>.$queue.<subject>, into a JetStream stream that keeps it
	// for OfflineQueueMaxAge, and the PUBACK or PUBREC waits for JetStream to
	// store it. It covers sessions held in this broker's memory and, with
	// PersistentSessions, sessions restored from the session store after a
	// restart or on another broker. Messages published straight onto NATS
	// rather than through a broker are not queued (ROADMAP.md).
	//
	// The queue also backs shared subscriptions: each gets a durable consumer
	// on the queue stream that only members with a connected client pull
	// from, so QoS 1 and 2 work goes to members that can take it and waits
	// when none can. Brokers sharing a SubjectPrefix must agree on whether the
	// queue is on, or a message can reach a group twice.
	OfflineQueue bool
	// DisableOfflineQueue turns the default offline queue off, for a
	// deployment that would rather not store every QoS 1 and 2 message and
	// accepts that a resumed session misses what was published while it was
	// away. Setting it together with OfflineQueue is an error.
	DisableOfflineQueue bool
	// CloseBehindClients disconnects a client that is 2048 messages behind, with DISCONNECT 0x97 (Quota
	// exceeded), when the broker has no offline queue to catch it up from (DisableOfflineQueue, or no
	// JetStream) and a QoS 1 or 2 message was about to be dropped. Without it the broker drops the message
	// and logs it, and the client is not told: the loss of a message the client was owed is invisible to
	// it. With it the loss is a lost connection, which the client sees, and its session is kept to resume
	// (SessionExpiryInterval permitting); messages dropped before the close, and those published while it
	// is away, are still lost, as without a queue they always are. It does not slow the publishers, and it
	// closes only the connection that fell behind, not the broker's NATS connection. QoS 0 loss never
	// closes a connection. It does nothing while the offline queue is on, where such a client catches up.
	CloseBehindClients bool
	// DurablePublish makes the PUBACK of a QoS 1 PUBLISH and the PUBREC of a QoS 2
	// one mean that the message is safe, not only that the broker has it. It
	// implies OfflineQueue (the broker fails to start without the queue) and is
	// an error together with DisableOfflineQueue.
	//
	// Without it the broker already waits for JetStream to store the queue's
	// copy, but then hands the live copy to its NATS connection and
	// acknowledges at once; nats.go writes that from another goroutine, so a
	// broker killed in the window loses a live copy it has told the client
	// about. With it the broker also waits for the NATS server to confirm it
	// has processed the live publish (a flush: a PING answered by a PONG) before
	// it acknowledges. So an acknowledged message is (1) in the queue stream,
	// to the stream's replication and storage, for OfflineQueueMaxAge, and
	// (2) has reached the NATS server's routing. If either step fails or takes
	// longer than five seconds the client gets PUBACK / PUBREC 0x83
	// Implementation specific error and may send the message again.
	//
	// It does not make delivery to a subscriber exactly-once or guaranteed:
	// live delivery is core NATS, at most once per hop, and a session whose
	// client is away gets the queued copy on resume. It covers only QoS 1 and
	// 2 publishes from clients. A Will Message has no client to refuse to, so
	// its queue copy is tried up to three times instead (a try that timed out
	// may have stored its copy, so a Will can be queued twice). Costs a NATS round trip per
	// such publish on top of the queue's; see the README for measurements.
	DurablePublish bool
	// DurablePublishDedupe stops the retry of a refused DurablePublish from leaving a second copy in the
	// queue. A QoS 1 or 2 PUBLISH refused with 0x83 after its queue copy was stored (the flush failed, or
	// the queue answered too late) is sent again by the client, and without this the queue then holds the
	// message twice and a session that was away is offered it twice. With it, the broker gives each such
	// publish a Nats-Msg-Id and a retransmission of it (same packet identifier, DUP set, same topic and
	// payload) carries the same one, so JetStream drops the second copy inside the queue stream's
	// duplicate window (the stream's Duplicates, 2 minutes unless the operator changed it).
	//
	// A retry after the window has passed, one on a session that did not survive (Clean Start), or one that
	// differs in topic or payload is queued as before, and a live copy is still sent again, so a subscriber
	// that was connected can still see a retried message twice, as QoS 1 allows. A publish that was
	// acknowledged is forgotten, so a later message that reuses the packet identifier is never mistaken
	// for a duplicate. Requires DurablePublish; it is an error without it.
	DurablePublishDedupe bool
	// NATSIngest lets a plain NATS client publish through a broker. A message
	// published on a live subject straight onto NATS has no queue copy, so a
	// session that was away never gets it and a shared subscription gets it
	// only through its NATS queue group, which can pick a member whose client is
	// away. With this option the broker also listens on
	// <SubjectPrefix>.$ingest.<subject>, in a NATS queue group of every broker
	// that has it on, so one broker takes each message and handles it as an
	// MQTT PUBLISH on <subject>: the Mqtt5-* headers give its QoS, Retain flag
	// and properties (as on a live copy, see the README), the topic is checked
	// as a PUBLISH's is (and one starting with $ is always refused), a retained
	// message is stored, a QoS 1 or 2 message gets
	// its queue copy, and the live copy goes out with the broker's id and queue
	// sequence. A request is answered once that is done: an empty reply carrying
	// Mqtt5-Msg-Id and, when queued, Mqtt5-Queue-Seq; or one carrying
	// Mqtt5-Ingest-Error with the reason, after which nothing was published.
	//
	// The Authorizer is not asked: a NATS client has no MQTT identity, and NATS
	// permissions on the ingest subject decide who may use it. Off by default,
	// and then nothing listens on the subject.
	NATSIngest bool
	// ReconnectCatchUp makes a connected session get the QoS 1 and 2 messages
	// published while this broker's NATS connection was down. Live delivery is
	// core NATS: a broker that has lost its connection has no interest on the
	// server, so a message another broker publishes meanwhile reaches none of its
	// sessions, and without this option nothing sends it later (only a session
	// that was away is replayed).
	//
	// With it, when the connection drops the broker stops handing its sessions
	// live QoS 1 and 2 copies, and when it is back each connected session reads
	// the offline queue from OfflineQueueRewind before the drop: what it has
	// not been sent goes out first, in stream order, then live delivery
	// resumes. The ids the session keeps stop it sending anything twice.
	//
	// It needs the offline queue (an error together with DisableOfflineQueue)
	// and covers only messages with a queue copy: those published through a
	// broker. It covers a drop of this broker's own connection, which nats.go
	// reports; a copy lost between two NATS servers of a cluster while both
	// brokers stay connected is not seen (ROADMAP.md). Costs one read of the
	// queue stream per connected session after each reconnect. Off by default.
	ReconnectCatchUp bool
	// OfflineQueueMaxAge is how long a queued message is kept. Defaults to 24
	// hours. A session away for longer loses what is older.
	OfflineQueueMaxAge time.Duration
	// OfflineQueueRewind is how far before a session's connection ended its
	// replay reaches for a queued message whose live copy never arrived (the
	// publisher stopped between storing the queue copy and publishing the live
	// one), and for one still on its way, when the session has no replay
	// cursor: one restored from a record written before the cursor existed, or
	// one whose cursor was dropped for its bounds (see the README). A session
	// with a cursor replays from it and finds such a message however long ago it
	// was stored. The session keeps the ids of what it delivered in that span so
	// that nothing is sent twice, so a longer rewind costs memory per session
	// and a read of that much of the queue stream on every resume. Defaults to
	// 10 seconds.
	OfflineQueueRewind time.Duration
	// OfflineQueueStorage selects file or memory storage for the queue
	// stream. Defaults to file storage.
	OfflineQueueStorage jetstream.StorageType
	// OfflineQueueReplicas is the JetStream replica count for the queue
	// stream. Defaults to 1.
	OfflineQueueReplicas int

	// Authenticator, when non-nil, decides whether a CONNECT is accepted. When
	// nil every connection is accepted, which is only appropriate on a trusted
	// network.
	Authenticator Authenticator
	// Authorizer, when non-nil, decides whether a session may publish to or
	// subscribe to a topic. When nil everything is permitted.
	Authorizer Authorizer
	// Auther, when non-nil, offers enhanced authentication: the AUTH-packet
	// exchange of MQTT-5.0 §4.12 for a CONNECT that names an Authentication
	// Method, and re-authentication on an established connection. When nil such a
	// CONNECT is refused with 0x8C (Bad authentication method).
	Auther Auther

	// ReceiveMaximum is advertised in CONNACK and bounds the client's in-flight
	// QoS 1 and QoS 2 publications. A client that has more QoS 2 PUBLISH packets
	// unacknowledged than this is sent DISCONNECT 0x93 and disconnected
	// (MQTT-5.0 §3.3.4). Defaults to DefaultReceiveMaximum.
	ReceiveMaximum uint16
	// MaximumQoS is the highest QoS the broker accepts in a PUBLISH and grants
	// on a SUBSCRIBE, advertised in CONNACK [MQTT-3.2.2-9]. Values above 2 are
	// clamped. Leave it nil for the default of 2; it is a pointer so that
	// Ptr(uint8(0)) can express a broker that only does QoS 0, which the zero
	// value would otherwise make unreachable.
	MaximumQoS *uint8
	// SharedQoS2 grants QoS 2 on a shared subscription ($share/...) when a
	// client asks for it; by default the SUBACK grants QoS 1, which
	// [MQTT-3.8.4-7] permits. Off until the exactly-once guarantee for a message
	// a shared member holds is complete (ROADMAP.md, "QoS 2 on shared
	// subscriptions"); turning it on earlier is for trials.
	SharedQoS2 bool
	// PublisherClockExpiry counts a live message's Message Expiry Interval from when the
	// publishing broker received it, not from when this broker did. The publisher stamps the
	// time (sub-second) on messages that carry an interval, so the wait spent in NATS between
	// brokers is no longer free and the remainder is not rounded up per hop. Off by default.
	//
	// It trusts the stamp, and so the clocks: a stamp in the future counts as no wait, but a
	// publisher with a clock running behind makes messages expire early, and anything able to
	// publish on the NATS subject can forge one. Enable it on every broker sharing the subjects,
	// on hosts with synchronised clocks, and where only brokers publish to them.
	PublisherClockExpiry bool
	// TopicAliasMaximum is advertised in CONNACK. Zero disables inbound topic
	// aliases; leave it unset for DefaultTopicAliasMaximum.
	TopicAliasMaximum *uint16
	// OutboundTopicAliases lets the broker send Topic Aliases to a client that
	// offered a Topic Alias Maximum in its CONNECT (MQTT-5.0 §3.3.2.3.4), up to
	// that maximum, least recently used first. It saves bandwidth on long topic
	// names that repeat. Off by default; a client that offers no maximum is
	// never sent one either way. A resend after a reconnect carries the full
	// name again, since the table belongs to the network connection.
	//
	// Deprecation notice: this is off by default today, and the default will
	// become on in the next major version (MIGRATION.md). Set
	// PinOutboundTopicAliases to choose now; leaving both unset logs one INFO
	// line per process.
	OutboundTopicAliases bool
	// PinOutboundTopicAliases fixes OutboundTopicAliases whatever the default
	// is, and beats OutboundTopicAliases when both are set: Ptr(false) sends
	// no alias after the default flips, Ptr(true) sends them today. nil leaves
	// the choice to OutboundTopicAliases and the default.
	PinOutboundTopicAliases *bool
	// EscapeNATSWildcards lets a topic contain '*', '>', spaces and the other
	// whitespace characters, which MQTT allows but a NATS subject cannot carry,
	// so the broker refuses them by default. When set, each is stored in the
	// NATS subject as '/' and a letter ("/s" for '*', "/g" for '>', "/p" for a
	// space, "/t", "/n", "/r", "/f" and "/d" for tab, line feed, carriage
	// return, form feed and DEL). The encoder never writes '/' before a letter,
	// so the escape is unambiguous, and a topic that has none of those
	// characters maps to the same subject as before. The cost is
	// interoperability: a NATS-native client sees the escaped form, and
	// nats-server's own MQTT listener, which refuses these topics, does not
	// read it. Decide before you have data: turning it on or off later leaves
	// existing retained, queued and in-flight messages under the other mapping.
	EscapeNATSWildcards bool
	// MaximumPacketSize caps an inbound packet. Defaults to
	// DefaultMaximumPacketSize.
	MaximumPacketSize uint32
	// ServerKeepAlive, when non-zero, overrides the client's requested Keep
	// Alive and is returned in CONNACK [MQTT-3.1.2-21].
	ServerKeepAlive uint16
	// ConnectTimeout bounds the wait for a CONNECT on a new connection.
	// Defaults to DefaultConnectTimeout.
	ConnectTimeout time.Duration
	// MaxSessionExpiry caps a requested Session Expiry Interval. Defaults to
	// DefaultMaxSessionExpiry.
	MaxSessionExpiry time.Duration
	// SessionSweepInterval is how often the broker discards detached sessions
	// whose Session Expiry Interval has passed, tearing down their NATS
	// subscriptions, and retained messages whose Message Expiry Interval has
	// passed. A session therefore outlives its interval by up to this
	// long, never less [MQTT-3.1.2-23]; a client that returns after the interval
	// but before the sweep still gets Session Present 0. Defaults to
	// DefaultSessionSweepInterval.
	SessionSweepInterval time.Duration
	// SessionCheckpointInterval is the longest a broker with PersistentSessions
	// goes between writing where a connected client stands (the offline-queue
	// position its replay would start from) to the session record, while that
	// changes. It bounds what a broker killed outright costs its sessions: the
	// broker that claims the record next replays from the last position written,
	// so a client is sent again at most what it was sent in this long, instead
	// of everything the queue holds (OfflineQueueMaxAge). The cost is one
	// key-value write per interval for each connection whose position moved.
	// Defaults to DefaultSessionCheckpointInterval; a negative value turns the
	// checkpoint off, and a killed broker's sessions then replay the whole queue.
	SessionCheckpointInterval time.Duration

	// Logger receives broker events. Defaults to slog.Default().
	Logger *slog.Logger
}

Options configures a Broker. The zero value is usable: it connects to nats.DefaultURL and listens on :1883.

Fields are only ever added to this struct, never removed or repurposed, so a composite literal with field names keeps compiling across versions.