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.
• 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.12.1

Images are published for linux/amd64 and linux/arm64 on every release, as :v0.12.1, :0.12.1, :0.11 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.12.1/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.12.1 -h for the full list.

As a binary:

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

As a library:

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

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)
	}
}

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 resumed session's replay also reaches back OfflineQueueRewind (10 seconds by default) before it was released, to catch a message whose live copy never arrived. 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), 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 (ROADMAP.md). A retry after a refusal can leave a second queue copy. Wills and QoS 0 are not covered.

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.

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: a record is removed only by a broker that publishes, cancels or drops it, so one left behind (see ROADMAP.md) stays until a broker adopts it. It is off by default because the Will's payload sits in clear text in that bucket, which therefore needs the access control of the session bucket, and because it needs JetStream. 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 4.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.

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.
• 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 at most. A QoS 2 request on a $share/ filter is answered with Granted QoS 1 in the SUBACK, which [MQTT-3.8.4-7] permits. Exactly-once delivery to one member of a group needs per-member ownership of a half-finished PUBREC/PUBREL exchange, and JetStream's unit of ownership is a consumer, not a queue-group member.
• No enhanced authentication (AUTH packets). A CONNECT carrying an Authentication Method is refused with 0x8C Bad authentication method, which is the conforming answer from a server that does not support it.
• The broker never sends topic aliases, though it accepts them from clients.
• 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 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. 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 deliberate choice here; see ROADMAP.md for the escaping option.

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()))
}

Index

• Constants
• Variables
• func Ptr[T any](v T) *T
• type Action
  • func (a Action) String() string
• type AuthRequest
• type AuthResult
• type Authenticator
• type AuthenticatorFunc
  • func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
• 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
• type ConnectError
  • func (e *ConnectError) Error() string
• type Options

Examples

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

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 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 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 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 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.

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

	// 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
	// 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.
	RestrictDollarTopics 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.
	DurableWills 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
	// reach it for twice this interval closes its client connections, and a
	// survivor waits about 4.5 intervals (plus the liveness ping timeout) before
	// adopting a silent broker's Wills, so a Will is never published while its
	// client's connection is open.
	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
	// 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, not Will Messages. Costs a NATS round trip per
	// such publish on top of the queue's; see the README for measurements.
	DurablePublish 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. 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

	// 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
	// TopicAliasMaximum is advertised in CONNACK. Zero disables inbound topic
	// aliases; leave it unset for DefaultTopicAliasMaximum.
	TopicAliasMaximum *uint16
	// 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.