mqttv5

package module
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

README

mqttv5

Go Reference CI License

An MQTT v5 client for Go: a low cost per message, an API that reads like Go, and QoS 1/2 sessions that survive reconnects and, with a file store, process restarts.

  • One package, standard library only. Persistence, WebSocket and codecs are optional submodules.
  • Reconnect, session resumption and re-subscription are built into every Client.
  • Messages arrive on a channel, a bounded queue or a callback, and are acknowledged when you say so.
  • Several brokers from one service: fail over between a cluster's nodes, or reach devices spread over independent clusters with one publish and one merged subscription (Multiple brokers).
  • Publishing can move to a pool of publish-only connections to the same broker: more QoS 1/2 publishes in flight than one connection allows, and none of those the members carry waiting behind a slow SubscribeCallback handler (Publish pool).
  • Decoding a PUBLISH of up to 64 KiB allocates nothing once pools are warm.
go get github.com/ashtonian/mqttv5

Requires Go 1.27 or later. Licensed under Apache 2.0.

Contents

Quick start

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/ashtonian/mqttv5"
)

func main() {
    ctx := context.Background()

    cli, err := mqttv5.New(
        mqttv5.WithBroker("mqtt://localhost:1883"),
        mqttv5.WithClientID("quickstart"),
    )
    if err != nil {
        log.Fatal(err)
    }
    if err := cli.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer cli.Disconnect(ctx)

    // Reconnects, session resumption and re-subscription happen underneath.
    msgs, _, err := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "sensors/#", QoS: 1}})
    if err != nil {
        log.Fatal(err)
    }

    err = cli.Publish(ctx, mqttv5.PublishOptions{Topic: "sensors/a1", QoS: 1, Payload: []byte("22.5")})
    if err != nil {
        log.Fatal(err)
    }

    m := <-msgs
    fmt.Printf("%s: %s\n", m.Topic, m.Payload)
    _ = m.Ack() // the broker gets its PUBACK once the message is acked
}

Run it against a local broker:

docker run -d -p 1883:1883 eclipse-mosquitto mosquitto -c /mosquitto-no-auth.conf
go -C examples run ./readme/quickstart

Why mqttv5

Compared with eclipse/paho.golang and its autopaho supervisor:

  • One client, supervisor built in. Reconnect, in-flight replay and re-subscription are always on; there is no second package to wire up.
  • A Go-shaped API. Channels and queues for delivery as well as callbacks, context.Context on every call, sentinel errors for errors.Is, and functional options instead of a large options struct.
  • A low cost per message. Decoding a PUBLISH of up to 64 KiB allocates nothing once pools are warm; a delivered message costs two allocations at QoS 0 (Performance).
  • Messages you can keep. A delivered *Message owns its topic, payload and properties, so it is safe to retain, and Ack may be called more than once. SubZeroCopy() skips the copy on hot paths.
  • Backpressure per subscription. A consumer that falls behind loses QoS 0 messages, never QoS 1 or 2: the broker's flow control holds those back, unless the subscription opts out with SubAutoAck or a DropOldest queue. A dropped message is acknowledged so the broker stops resending it.
  • Several brokers, two distinct patterns. Failover (WithBrokers) and parallel sessions (ClientGroup), each its own API, and they compose (Multiple brokers).
  • Publishing apart from receiving. WithPublisherPool adds publish-only connections to the same broker, each with a QoS 1/2 window of its own; subscriptions stay on the main connection, so a slow SubscribeCallback handler holds up no publish a member carries (Publish pool).
  • Durable publishing. QueuePublisher with queue/file and store/file keeps publishing through outages and survives a process crash without publishing an in-flight message twice.
  • Typed payloads. One Encoding on the client, the payload type per call: cli.SubscribeAs[Reading](…) delivers decoded values and cli.PublishValue(…) encodes any value and declares the MQTT 5 Content Type. JSON and msgpack Encodings ship as submodules, so the core stays standard library only.
  • The MQTT v5 feature set, tested against real brokers. Shared subscriptions, Subscription Identifiers, topic aliases in both directions, session expiry and resumption, retained messages, the Will and its properties, server redirects, enhanced authentication (CONNECT and mid-session, §4.12), and the broker's CONNACK limits enforced before a packet is sent. The conformance suite runs against mosquitto and EMQX on every change, and nightly against HiveMQ CE, including one that announces tight CONNACK limits.
  • WebSocket without a core dependency. transport/ws plugs in through WithDialFunc (WebSocket).

Examples

examples/ is one module of runnable programs. They read the broker from MQTT_BROKER (default mqtt://127.0.0.1:1883), the multi-broker ones from the variables in examples/README.md.

Path Shows
examples/readme/quickstart The quick start above: connect, channel subscribe, publish
examples/subscribe Channel, queue and callback subscriptions, and what each does with a consumer that falls behind
examples/typed SubscribeAs[T] and PublishValue with the JSON Encoding: two payload types on one client, and a payload that does not decode
examples/request MQTT 5 request/response: Response Topic, Correlation Data, a shared responder
examples/durable QueuePublisher on queue/file with store/file: publishing through outages and restarts
examples/reconnect Full lifecycle callback set (Up / Down / ConnectError / ReconnectAttempt) surviving a broker restart
examples/failover WithBrokers failover; ConnackInfo.Broker names the broker reached
examples/pool WithPublisherPool: QoS 1 throughput with and without a pool
examples/group ClientGroup: a central service sending shadow updates to devices on either of two clusters, acknowledged in the application by the devices' replies
examples/bridge At-least-once bridge between two brokers, surviving outages and restarts
examples/shared Workers sharing a subscription ($share)
examples/ws WebSocket transport — WithDialFunc(ws.DialFunc(opts))
examples/stats Client.Stats() snapshot — bridge into Prometheus / OTel / expvar
examples/oauth WithConnectPacketBuilder rotating an OAuth bearer per CONNECT
examples/disconnect DisconnectWith carrying ReasonCode + ReasonString + SessionExpiry override 
docker run -d -p 1883:1883 eclipse-mosquitto mosquitto -c /mosquitto-no-auth.conf
go -C examples run ./readme/quickstart

Modules

The core uses the standard library only. Each optional submodule has its own go.mod, so the core never pulls in its dependencies.

Submodule Import Purpose
core github.com/ashtonian/mqttv5 Client, supervisor, options, in-memory queue
JSON codec github.com/ashtonian/mqttv5/codec/json Encoding for the typed calls (stdlib only)
msgpack codec github.com/ashtonian/mqttv5/codec/msgpack Encoding via vmihailenco/msgpack/v5
File session store github.com/ashtonian/mqttv5/store/file Crash-safe in-flight QoS 1/2 state
File publish queue github.com/ashtonian/mqttv5/queue/file Crash-safe outbound publish queue for QueuePublisher
WebSocket transport github.com/ashtonian/mqttv5/transport/ws ws:// and wss:// — WithDialFunc(ws.DialFunc(ws.DialOpts{...}))

Every module is released with the same version tag (store/file/vX.Y.Z alongside vX.Y.Z); install a submodule at the version of the core you use:

go get github.com/ashtonian/mqttv5@latest github.com/ashtonian/mqttv5/store/file@latest

store/file and queue/file bring in internal/filedb, the bbolt layer they share, at the same version.

Publishing

err := cli.Publish(ctx, mqttv5.PublishOptions{
    Topic:   "sensors/a1/temp",
    QoS:     1,
    Payload: []byte(`{"temp":22.5}`),
})

When Publish returns depends on the QoS:

QoS Publish returns
0 once the packet is queued for the connection's writer; with WithPublishMode(PublishWaitForFlush), once it is written
1 when the broker's PUBACK arrives
2 when the broker's PUBCOMP arrives

ctx bounds the call. While the client is disconnected Publish returns ErrNotConnected; with a publish pool that means no member is connected either, since publishing goes on through the members while the main connection reconnects. A QueuePublisher queues through outages instead. A QoS 1/2 message accepted before a drop belongs to the session: it is resent when the session resumes, and the caller stays blocked until the broker acknowledges it. A refusal (reason code 0x80 or above) is returned as a *ReasonCodeError.

Write path
  • On a TCP or Unix connection with nothing queued, a publish that waits for its write (QoS 1/2, or QoS 0 with PublishWaitForFlush) writes on its own goroutine: no hand-off, and a QoS 0 header and payload go out in one writev without the payload being copied.
  • When publishers overlap, their packets queue for the connection's writer goroutine, which writes them together; WithWriteBatch(n) coalesces them into one writev. Fire-and-forget QoS 0 always goes through the writer.
  • If ctx ends during a write, the writer goroutine finishes the packet, so it may still reach the broker after Publish returned ctx's error.
Broker limits

Every publish, subscribe and unsubscribe is checked against the CONNACK before anything is written:

  • QoS above Maximum QoS → ErrQoSNotSupported, or the QoS is lowered with WithQoSDowngrade();
  • retain with Retain Available = 0 → ErrRetainNotSupported;
  • a packet above Maximum Packet Size → ErrPacketTooLarge.

At most Receive Maximum QoS 1/2 publishes are in flight; further ones wait, bounded by ctx (Stats().SendQuota shows the headroom). Stored messages that a reconnect to a stricter broker rules out complete with the same errors. Client.ServerInfo() returns everything the broker granted.

Publish pool: WithPublisherPool

A broker lets each connection have at most its Receive Maximum QoS 1/2 publishes unacknowledged (by default 20 for mosquitto, 32 for EMQX, 10 for HiveMQ CE) and handles each connection's packets in order, so one connection carries at most about Receive Maximum QoS 1/2 publishes per round trip, however many goroutines call Publish. The pool opens n more connections to the same broker, publish-only, each with a window of its own, and spreads Publish calls over them. Only the main connection subscribes: subscriptions, the Will, a persistent session (WithSessionExpiry, WithStore), the lifecycle callbacks and QueuePublisher stay on it, and a member leaves nothing on the broker when it disconnects.

flowchart LR
    pub["Publish"] --> route{"routing:<br/>round robin, or<br/>hash of the topic"}
    route --> m1["pool member 1"]
    route --> m2["pool member 2"]
    route --> m3["pool member n"]
    route -. "no member can send" .-> main["main connection"]
    m1 & m2 & m3 & main --> broker[("broker")]

cli, _ := mqttv5.New(
    mqttv5.WithBroker("mqtts://mqtt.example.com:8883"),
    mqttv5.WithClientID("telemetry-gw"),
    mqttv5.WithPublisherPool(4), // members telemetry-gw-pub-1 … -4
    mqttv5.WithPublisherPoolRouting(mqttv5.PoolRoutingHashByTopic),
)

Publish is called as it is without a pool.

  • When it helps. When the round trip limits throughput: several goroutines publishing QoS 1 or 2 to a broker across a WAN, or to one slow to acknowledge. A nearby broker is limited by its own processing instead, and a pool adds little; QoS 0 has no window to widen. examples/pool measures the difference against your broker.
  • Routing. Each Publish goes out on one connection. Round robin (the default) takes the next member each call and may reorder a topic; PoolRoutingHashByTopic keeps each topic on one member, and so in order while that member stays connected.
  • Failure. A member that cannot send (not connected, or its write queue full under WriteDropNewest) passes the message to the next; when none can, the main connection sends it, counted in Stats().PoolFallbacks. A broker refusal or ctx ending is returned as is, so a message never goes out twice. A member whose first CONNECT fails keeps retrying in the background; Stats().PoolConnected counts the members connected.
  • Guarantees. Members have no session that outlives their connection: each connects with Session Expiry 0 (and Clean Start = 1 the first time). A QoS 1/2 publish interrupted by a member's reconnect follows the session-loss policy — sent again, possibly duplicated, so QoS 2 is not exactly-once across it. Publish what must resume exactly on a connection with a session: a QueuePublisher always uses the main connection.
  • The broker's view. Each member is a client of its own to the broker: n more connections against its connection limits, with ClientIDs <ClientID>-pub-1 … -n that any ACL or policy keyed on the ClientID must admit (WithPublisherPoolClientIDFn changes them). A client without a ClientID leaves its members' to the broker too.
  • No Local. A pooled publish comes from a member's ClientID, so a No Local subscription on the main connection receives it; only a publish that fell back to the main connection is held back. Mark your own messages (with a user property, say) and skip them in the handler instead.
  • Brokers. Members use the client's broker list — SetBrokers replaces it for them too — and each starts at a random entry, so over a cluster the pool spreads across its nodes. Each member follows the server redirects its own broker sends.
  • Lifetime. Members connect with the client and disconnect with it, never on their own: an idle member stays connected and sends keepalive pings. When the main connection drops, the members stay connected and publishing goes on. Disconnect refuses new publishes with ErrClosed, waits, bounded by its ctx, for those in flight on members (a member's session ends with its connection, so nothing could finish them later), then disconnects the members. A publish its ctx cuts short returns ErrClosed, and the broker may or may not have it; Disconnect then returns ctx's error.
  • Client-wide calls. Reauthenticate re-authenticates every connected member along with the main connection, so the Authenticator must be safe for concurrent use. Stats() counts the members' connections and publishes with the main connection's. Connected(), AwaitConnection and the lifecycle callbacks report the main connection only.
  • Receiving and publishing apart. A connection reads the acknowledgements of its own publishes on the same loop that runs SubscribeCallback handlers, SubOnDrop hooks and typed decoding, so on the main connection a slow handler holds up publishes too: with 100 messages queued for a 10 ms handler, a QoS 1 publish waits about a second. A member reads its own acknowledgements and never waits on them, so its publishes don't wait — though a publish that falls back to the main connection does. Channel and queue subscriptions hand messages to their consumer without waiting for it; only their typed decoding and SubOnDrop / SubOnDecodeError hooks run on the loop.
  • A pool of one adds no window, but is enough to keep publishes from waiting behind the subscriptions' handlers.

Members are configured from the parent: brokers, dialing, TLS, credentials, authenticator, CONNECT properties, timeouts, buffer sizes, write path, backoff, and the publish behaviour (WithQoSDowngrade, WithSessionLossPolicy, WithOutboundTopicAliases, WithLenientDecoding, WithFollowServerRedirects).

Subscribing

Three delivery shapes. Each takes []TopicFilter, so several filters go out in one SUBSCRIBE.

Channel — manual ack, ordered flush
msgs, token, err := cli.Subscribe(ctx,
    []mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}},
    mqttv5.SubBuffer(256),
)
for m := range msgs {
    handle(m)
    _ = m.Ack() // PUBACK released in §4.6 arrival order
}
_ = cli.Unsubscribe(ctx, token) // closes msgs

A consumer that falls behind loses QoS 0 messages, never QoS 1 or 2. Once the channel holds SubBuffer messages, an incoming QoS 0 message is acked and dropped; observe drops via SubOnDrop(...). QoS 1 and 2 messages have room for WithReceiveMaximum more (default 256): as many as the broker sends before the consumer acks one, so the broker holds the rest back. Two exceptions: SubAutoAck() acks on receipt, which gives up that flow control, so QoS 1 and 2 messages are dropped as QoS 0 ones are; and messages still unread from a session the broker discarded take room of their own.

Queue — bounded, optional DropOldest
q, _, _ := cli.SubscribeQueue(ctx,
    []mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}},
    mqttv5.SubMaxQueueSize(10_000),
    mqttv5.SubDropPolicy(mqttv5.DropOldest), // keeps freshest 10k
)
for {
    m, ok := q.Dequeue(ctx)
    if !ok {
        break
    }
    handle(m)
    _ = m.Ack()
}

Queues hold at most DefaultMaxSubscribeQueueSize (65,536) messages unless WithMaxSubscribeQueueSize / SubMaxQueueSize say otherwise; mqttv5.UnboundedQueue removes the cap. When full, DropNewest (the default) acks and drops an incoming QoS 0 message, while QoS 1 and 2 messages have the same room beyond the cap as on a channel; DropOldest evicts the queue head, whatever its QoS, and acks it before enqueueing.

Only the queue variant supports DropOldest — channels can't peek-and-pop without racing the consumer.

Callback — sync, auto-ack
cli.SubscribeCallback(ctx,
    []mqttv5.TopicFilter{{Topic: "ctrl/+", QoS: 0}},
    func(m *mqttv5.Message) {
        // Runs on the read goroutine — MUST be non-blocking: the
        // acks of this client's own publishes wait behind it.
        process(m)
        // Ack auto-fires after return.
    },
)

Pass mqttv5.SubZeroCopy() as a trailing option to skip the copy; the message's fields are then valid only until the callback returns. The handler must not call Disconnect, which waits for the read goroutine; start it on another goroutine. While a handler runs, everything else on the connection waits behind it: other subscriptions' messages and the acknowledgements of this client's own publishes (a publish pool takes publishing off the connection). The keep-alive does not take that wait for a dead broker. Work that may take long belongs on a channel or queue subscription.

Durable consumers

A consumer with a session the broker keeps (a fixed ClientID, WithCleanStart(false), and a WithSessionExpiry longer than its downtime) gets the messages that arrived while it was down. The broker sends them right after the CONNACK, before Connect has returned, so subscribe before Connect: Subscribe then puts the subscription in place at once and its SUBSCRIBE goes out first on the connection. A message no subscription matches is acknowledged and dropped, so a Subscribe after Connect races the broker and loses whatever arrives first; such messages are counted in Stats().InboundUnrouted, and the first on each connection is logged at Warn.

cli, _ := mqttv5.New(
    mqttv5.WithBroker(url),
    mqttv5.WithClientID("billing-worker-1"),
    mqttv5.WithCleanStart(false),        // resume the session after a restart
    mqttv5.WithSessionExpiry(24*60*60),  // the broker keeps it a day
)
msgs, _, _ := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "invoices/#", QoS: 1}})
if err := cli.Connect(ctx); err != nil {
    log.Fatal(err)
}
for m := range msgs { // first what arrived while the worker was down
    handle(m)
    _ = m.Ack()
}

Typed publish / subscribe

The client gets one Encoding; each call names its payload type, so one client handles any number of them.

import jsoncodec "github.com/ashtonian/mqttv5/codec/json"

type Reading struct {
    Device string
    Temp   float64
}

type Command struct {
    Op string
}

cli, _ := mqttv5.New(
    mqttv5.WithBroker(url),
    mqttv5.WithEncoding(jsoncodec.Encoding{}), // one Encoding, any payload type
)

// Subscribed before Connect, so nothing the broker sends is missed.
readings, _, _ := cli.SubscribeAs[Reading](ctx,
    []mqttv5.TopicFilter{{Topic: "sensors/#", QoS: 1}})
commands, _, _ := cli.SubscribeAs[Command](ctx,
    []mqttv5.TopicFilter{{Topic: "devices/+/cmd", QoS: 1}})
if err := cli.Connect(ctx); err != nil {
    log.Fatal(err)
}
defer cli.Disconnect(ctx)

// Declares Content Type application/json.
_ = cli.PublishValue(ctx, mqttv5.PublishOptions{Topic: "sensors/a1", QoS: 1},
    Reading{Device: "a1", Temp: 22.5})

for {
    select {
    case m, ok := <-readings:
        if !ok {
            return
        }
        fmt.Println(m.Topic, m.Value.Temp)
        _ = m.Ack()
    case m, ok := <-commands:
        if !ok {
            return
        }
        fmt.Println(m.Topic, m.Value.Op)
        _ = m.Ack()
    }
}
  • Subscribing. SubscribeAs[T], SubscribeQueueAs[T] and SubscribeCallbackAs are Subscribe, SubscribeQueue and SubscribeCallback with each payload decoded into a T, with the same buffering, bounds, drop policy, refusals and closing. A *TypedMessage[T] carries the decoded Value beside the *Message, which is acked as usual. Payloads are decoded on the connection's read goroutine as they arrive, straight into the delivered message.
  • Publishing. PublishValue encodes any value and publishes it as Publish does, declaring the Encoding's Content Type unless PublishOptions.ContentType is set; PublishOptions.Payload must be empty.
  • A payload that does not decode is not delivered: it is acked, so the broker does not send it again, logged at Warn, counted in Stats().DecodeFailures, and handed with the error to SubOnDecodeError if the subscription set one.
  • Several encodings. SubEncoding decodes one subscription with another Encoding than the client's. A ClientGroup member uses its own, so a group can join brokers whose payloads differ.
  • Groups and queues. ClientGroup.SubscribeAs, ClientGroup.SubscribeQueueAs and ClientGroup.PublishValue decode and encode with each member's Encoding, and need one on every member; QueuePublisher.PublishValue encodes before the message is queued.
  • Without an Encoding every typed call returns ErrNoEncoding before anything is sent.

codec/json and codec/msgpack provide an Encoding each. For protobuf, Cap'n Proto, FlatBuffers or a format of your own, implement the three methods (ContentType, Marshal, Unmarshal); Unmarshal must not keep its input, which on a SubZeroCopy subscription is the network frame. The typed calls are generic methods (Go 1.27), and an interface cannot list a generic method: code that hides *Client behind an interface lists Subscribe and decodes itself.

Request / response

MQTT 5 carries the reply address in the request: a Response Topic and Correlation Data, which the responder copies into its reply.

// Responder: reply on each request's Response Topic, with its
// Correlation Data. A shared subscription spreads requests over instances.
reqs, _, _ := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "$share/pricing/requests/price", QoS: 1}})
for m := range reqs {
    if rt := m.Properties.ResponseTopic(); rt != "" {
        _ = cli.Publish(ctx, mqttv5.PublishOptions{
            Topic:           rt,
            CorrelationData: m.Properties.CorrelationData(),
            QoS:             1,
            Payload:         price(m.Payload),
        })
    }
    _ = m.Ack()
}

// Requester: one reply subscription serves every request in flight.
replyTo := "replies/" + cli.ClientID()
replies, _, _ := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: replyTo, QoS: 1}})
_ = cli.Publish(ctx, mqttv5.PublishOptions{
    Topic:           "requests/price",
    QoS:             1,
    Payload:         []byte("A-100"),
    ResponseTopic:   replyTo,
    CorrelationData: []byte("req-1"), // matches the reply to this call
})

A requester with several requests in flight matches each reply to its call by Correlation Data; examples/request keeps the pending calls in a map, each with its own timeout. Brokers that assign reply topics announce a prefix in ConnackInfo.ResponseInformation when asked with WithRequestResponseInformation(true).

Durable publishing with QueuePublisher

QueuePublisher decouples the caller from broker availability:

  • Publish returns as soon as the message is stored in the queue.
  • A drain goroutine keeps up to WithQueueWindow messages in flight (default 32), in queue order, whenever the client is connected, and removes each from the queue once the broker has accepted it. Throughput approaches window ÷ round-trip time, the window capped by the broker's Receive Maximum (20 on mosquitto): against a broker that allows the whole window, at 25 ms, about 1,200 msg/s with the default window and 600 with a window of 16, against 38 msg/s one at a time (BenchmarkQueuePublisherRTT).
import (
    qfile "github.com/ashtonian/mqttv5/queue/file"
    sfile "github.com/ashtonian/mqttv5/store/file"
)

st, _ := sfile.Open("/var/lib/myapp/session")
cli, _ := mqttv5.New(mqttv5.WithBroker(url), mqttv5.WithClientID("dev-1"), mqttv5.WithStore(st))
q, _ := qfile.Open("/var/lib/myapp/outbound")
pub, _ := mqttv5.NewQueuePublisher(cli, q,
    mqttv5.WithQueueMaxSize(1_000_000),
    mqttv5.WithQueueTTL(24*time.Hour),
    mqttv5.WithDeadLetter(func(e mqttv5.QueueEntry, err error) {
        log.Printf("dropped %s: %v", e.Publish.Topic, err)
    }),
)
defer pub.Close(ctx)

_ = pub.Publish(ctx, mqttv5.PublishOptions{Topic: "logs", Payload: data, QoS: 1})

Failures. A refusal that cannot pass — PUBACK/PUBREC 0x87 Not authorized, 0x90 Topic Name invalid, 0x95 Packet too large, 0x99 Payload format invalid, 0x9A Retain not supported, 0x9B QoS not supported, or the same verdict from the broker's CONNACK limits — goes to the dead-letter callback and the message is removed (WithQueueClassifier replaces the rule). Anything else, such as 0x97 Quota exceeded, is retried after WithQueueRetryBackoff, behind the messages that follow it, so one bad message never blocks the queue. An expired message is dead-lettered with ErrMessageExpired instead of sent; one that is sent carries the lifetime it has left as its Message Expiry Interval.

One exchange per message. A slow broker never causes a second exchange for the same message. Each message's ID is stored with its exchange in the client's session (WithStore), and the message leaves the queue before the session forgets the exchange. With queue/file and store/file, a restarted process continues the exchanges its predecessor had in flight — QoS 1 resent with DUP=1, QoS 2 from where it stopped — rather than publishing those messages again. If removing a finished message from the queue fails, the session keeps the exchange's record and the removal is retried with WithQueueRetryBackoff (or by the next process) without publishing the message again. A message is published again as new only if the broker lost the session; enable WithQueueIdempotencyKey to add the message ID as the mqttv5-msg-id user property so consumers can drop that copy.

Bounds. WithQueueMaxSize is enforced atomically by the queue, so concurrent producers cannot exceed it. At the bound DropNewest (default) returns ErrQueueFull; DropOldest evicts the oldest messages not yet being published and dead-letters them with ErrQueueFull. Constructing a QueuePublisher never removes anything.

QoS 0 is rejected (ErrQoS0NotQueueable), as is a message that could never be sent (invalid topic, Topic Alias): the error comes from Publish, not a dead letter later. The publisher copies the message, so the caller may reuse its buffers.

A QueuePublisher publishes on the client's main connection, never through a publish pool: its exchanges are recorded in that connection's session, which is what lets them continue after a reconnect or restart.

Use mqttv5.NewMemoryPublisherQueue() for in-process buffering without crash safety. A custom PublisherQueue can be checked against the contract with github.com/ashtonian/mqttv5/queue/queuetest.

Authentication

WithCredentials(user, pass) sends a fixed username and password. For credentials that change, rotate them per connection attempt or re-authenticate the live connection; WithAuthenticator adds MQTT v5 enhanced authentication (challenge/response such as SCRAM).

Credentials per connection attempt

WithConnectPacketBuilder(fn) runs immediately before each CONNECT is serialised. Use it to refresh an OAuth token, fetch a SigV4-signed CONNECT credential, or rotate any other per-attempt secret. The context is bounded by ConnectTimeout.

mqttv5.WithConnectPacketBuilder(func(ctx context.Context, opts *mqttv5.ConnectOptions) error {
    tok, err := oauth.FetchToken(ctx)
    if err != nil {
        return err // fails this attempt; supervisor retries after backoff
    }
    opts.Username = "service-account"
    opts.Password = []byte(tok)
    return nil
}),

WithOnConnectError reports each failed attempt with its error. See examples/oauth.

Re-authentication without reconnecting

Client.Reauthenticate(ctx) drives MQTT v5 re-authentication (§4.12) on the live connection — no reconnect, no QoS-state churn:

  1. Sends an AUTH 0x19 carrying a fresh Authenticator.Begin(ctx) payload.
  2. Services any broker challenges via Continue.
  3. Returns when the broker concludes with 0x00 Success.

Use it on a long-lived connection whose bearer token (OAuth, JWT) expires: start a timer from the token's lifetime and call Reauthenticate ahead of expiry.

// e.g. 30s before the JWT `exp`:
if err := cli.Reauthenticate(ctx); err != nil {
    // ErrReauthRejected → broker refused the new credential;
    // the supervisor is already reconnecting with a fresh CONNECT.
}

Behaviour:

  • ctx bounds the whole operation, including the token fetch in Begin.
  • Calls are single-flighted per connection.
  • A broker rejection returns ErrReauthRejected and tears the connection down, so the supervisor reconnects through the normal CONNECT path.
  • WithOnReauthenticated observes every successful refresh in one place.
  • With a publish pool, every connected member re-authenticates at the same time; a member's failure is joined into the returned error, and a member that is not connected presents a fresh credential when it connects.
Mutual authentication

For mechanisms with mutual authentication (e.g. SCRAM), an Authenticator may also implement the optional ServerFinalVerifier interface (VerifyServerFinal([]byte) error).

The client hands it the server's concluding AuthenticationData — the CONNACK on connect, the AUTH 0x00 on re-auth — so it can verify the server proved knowledge of the credential.

A verification failure aborts the connect, or fails Reauthenticate and tears the connection down.

WebSocket

import (
    "github.com/ashtonian/mqttv5"
    "github.com/ashtonian/mqttv5/transport/ws"
)

cli, _ := mqttv5.New(
    mqttv5.WithBroker("wss://broker.example.com/mqtt"),
    mqttv5.WithDialFunc(ws.DialFunc(ws.DialOpts{TLSConfig: tlsCfg})),
)

Without TLSConfig, wss:// verifies the broker against the system roots with the URL's host as server name, the same default as mqtts://. ws:// and wss:// URLs need WithDialFunc; New rejects them otherwise.

Multiple brokers

One connection reaches one broker. Services outgrow that for two reasons, and each has an API of its own:

  • No single broker may take the service down. Brokers that carry the same data — the nodes of a cluster, or replicas — are interchangeable, so one client connects to whichever is up: failover.
  • The data is spread over brokers that do not share it. Devices connect to whichever of two independent clusters is up or nearer, regions or tenants have brokers of their own, or a fleet is moving from one broker to another. A central service needs a session on each: one publish reaches all of them and one subscription hears from all of them: parallel sessions, as in central service over independent clusters.

Which one fits depends on how the brokers relate to each other:

Pattern API Connections The brokers
Failover WithBrokers(urls...) one at a time are interchangeable: nodes of a cluster, or replicas
Parallel sessions NewClientGroup(members, opts...) one per member, all live are independent: each holds its own data 
flowchart TD
    same{"Do the brokers hold<br/>the same data?"}
    same -- "yes: a cluster or replicas" --> failover["Failover<br/>WithBrokers"]
    same -- "no: devices or data split across them" --> standby{"Does each one<br/>have a standby?"}
    standby -- no --> group["Parallel sessions<br/>ClientGroup"]
    standby -- yes --> regional["Regional HA<br/>ClientGroup, members with several Brokers"]

Publishing faster than one connection allows is a separate question, with the same answer for one broker or many: a publish pool opens more connections to the brokers the client already uses.

They compose: a ClientGroup member is a full Client, so it can fail over between brokers of its own and carry its own publish pool (What combines). The recipes build on them: a central service over independent clusters, a bridge that loses nothing, quorum writes, sharding with fan-in, a durable outbox, brokers from service discovery, and consumers scaled out with shared subscriptions.

Failover: WithBrokers

One client, one session identity, one connection at a time, moving between brokers that carry the same data.

flowchart LR
    client["Client ingest-7"] == "connected" ==> b1[("mqtt-1")]
    client -. "after a drop" .-> b2[("mqtt-2")]
    client -. "then" .-> b3[("mqtt-3")]

cli, _ := mqttv5.New(
    mqttv5.WithBrokers(
        "mqtts://mqtt-1.example.com:8883",
        "mqtts://mqtt-2.example.com:8883",
        "mqtts://mqtt-3.example.com:8883",
    ),
    mqttv5.WithClientID("ingest-7"),
    mqttv5.WithOnConnectionUp(func(info mqttv5.ConnackInfo) {
        log.Printf("connected to %s, session present: %v", info.Broker, info.SessionPresent)
    }),
    mqttv5.WithOnReconnectAttempt(func(attempt int, broker string) {
        log.Printf("reconnect attempt %d: %s", attempt, broker)
    }),
)
  • Connect. Each client starts at a random entry, so a fleet that restarts together spreads over the list. Connect tries each broker once and fails only when none accepts; its error joins every broker's, each naming its URL, and errors.Is sees them all. With WithRetryInitialConnect the client keeps trying in the background instead.
  • A drop. After the reconnect backoff the supervisor dials the next entry, so the first attempt leaves the broker that just failed, and each further attempt moves on one. The client stays on whichever broker accepts it; ConnackInfo.Broker, in OnConnectionUp and ServerInfo(), names it.
  • The session. Nodes of a cluster share sessions: the next one answers Session Present = 1, unacknowledged QoS 1/2 messages are resent with DUP set (QoS 2 stays exactly-once), and subscriptions stay in place. An independent broker has only the session it kept itself for this ClientID: usually none (Session Present = 0), so every subscription is subscribed again and unacknowledged publishes follow WithSessionLossPolicy — sent again as new messages (the default; a consumer may see a duplicate) or failed with ErrSessionLost. If it kept an older one from an earlier connection, that one resumes, messages it queued since included.
  • Callers. A Publish waiting for its acknowledgement keeps waiting across the failover, bounded by its ctx. While no broker is connected, Publish returns ErrNotConnected; a QueuePublisher queues.
sequenceDiagram
    participant C as Client
    participant A as mqtt-1
    participant B as mqtt-2
    C->>A: CONNECT
    A-->>C: CONNACK
    C->>A: PUBLISH QoS 1, id 7
    A--xC: connection lost before the PUBACK
    Note over C: OnConnectionDown, backoff,<br/>OnReconnectAttempt(1, mqtt-2)
    C->>B: CONNECT, Clean Start 0
    alt mqtt-2 shares the session (cluster)
        B-->>C: CONNACK, Session Present 1
        C->>B: PUBLISH QoS 1, id 7, DUP
        B-->>C: PUBACK 7
    else mqtt-2 is independent
        B-->>C: CONNACK, Session Present 0
        C->>B: SUBSCRIBE, every subscription
        C->>B: PUBLISH QoS 1, new id (SessionLossRepublish)
        B-->>C: PUBACK
    end
    Note over C: Publish returns
Changing brokers at runtime

A broker can send the client elsewhere (§4.11): a CONNACK or DISCONNECT with reason 0x9D Server moved or 0x9C Use another server and a Server Reference. WithOnServerRedirect reports each one; WithFollowServerRedirects() acts on it — a move replaces the broker list, a temporary redirect applies to the next connection attempt only. A reference that is a bare host or host:port keeps the current URL's scheme and path.

cli, _ := mqttv5.New(
    mqttv5.WithBroker("mqtts://broker-a.example.com:8883"),
    mqttv5.WithFollowServerRedirects(),
    mqttv5.WithOnServerRedirect(func(r mqttv5.ServerRedirect) {
        log.Printf("redirected to %s (permanent=%v)", r.Reference, r.Permanent())
    }),
)

Client.SetBrokers(urls...) replaces the list from your side, for a publish pool's connections too, from service discovery for example (recipe). The live connection stays where it is; the next connection attempt uses the new list.

Parallel sessions: ClientGroup

N sessions to N brokers, each member treated as itself:

  • a central service for devices split across independent clusters: every command to all of them, every reply from any of them (recipe),
  • a migration from one broker to another, publishing to both until the devices have moved,
  • bridges between independent brokers,
  • multi-tenant SaaS with per-broker credentials.
flowchart LR
    group["ClientGroup"] --> emea["member emea<br/>Client, own session"] --> b1[("emea broker")]
    group --> apac["member apac<br/>Client, own session"] --> b2[("apac broker")]

Members never stand in for each other: if your brokers are interchangeable for the same data, use WithBrokers on a single Client. Pointed at one cluster, a group's members would each subscribe to the same messages, so a merged subscription to a filter that is not a shared subscription would deliver every message once per member; for publish throughput into one broker or cluster, use a publish pool. A member can have several Brokers of its own to fail over between (regional HA).

Construction takes a GroupMember list plus group-level options:

g, _ := mqttv5.NewClientGroup(
    []mqttv5.GroupMember{
        {
            Brokers: []string{"mqtts://emea.example.com:8883"},
            Name:    "emea",
            Opts:    []mqttv5.Option{mqttv5.WithCredentials("emea-svc", []byte(token1))},
        },
        {
            Brokers: []string{"mqtts://apac.example.com:8883"},
            Name:    "apac",
            Opts:    []mqttv5.Option{mqttv5.WithCredentials("apac-svc", []byte(token2))},
        },
    },
    mqttv5.WithGroupSharedOpts(
        mqttv5.WithClientID("fleet"),
        mqttv5.WithKeepAlive(30),
    ),
    mqttv5.WithGroupPublishPolicy(mqttv5.GroupPublishBroadcast),
)

GroupMember.Opts applies after WithGroupSharedOpts, so per-member auth / TLS / ClientID / callbacks win; brokers come only from GroupMember.Brokers, and NewClientGroup rejects WithBroker or WithBrokers in either. Member names default to member-N (1-based) when unset. A member whose ClientID another member already uses gets the first free -N suffix (-2, -3, …); members without one have it assigned by their broker.

Publish returns every member's outcome ([]GroupResult) and an error decided by the policy below:

Publish policy Behaviour Use case
GroupPublishBroadcast (default) Publishes to every member at once and returns when each has answered or ctx ends. Fails with a *GroupError when fewer than WithGroupSuccess members succeed. Bridge / mirror / quorum — members carry different data
GroupPublishRoundRobin One member per call, the next each time. Spread publishes over independent brokers whose consumers subscribe to all of them
GroupPublishHashByTopic FNV-1a(topic) → member. Per-topic ordering. Shard topics across independent brokers (sharding with fan-in)

RoundRobin and HashByTopic send each message once. They move to the next member only when the chosen one could not send it at all (not connected); a broker refusal or the ctx ending is returned as is, so a message never goes out on two members.

Success policy (WithGroupSuccess) Broadcast Publish and Subscribe succeed when
GroupSuccessAll (default) every member succeeded
GroupSuccessAny at least one member succeeded
GroupSuccessQuorum(n) at least n members succeeded 
results, err := g.Publish(ctx, mqttv5.PublishOptions{Topic: "events/x", QoS: 1, Payload: p})
for _, r := range results {
    if r.Err != nil {
        log.Printf("%s: %v", r.Member, r.Err) // errors.Is(err, mqttv5.ErrNotAuthorized) works on the GroupError too
    }
}

Subscribe subscribes every member and merges their messages into one channel or queue; Ack goes back to the member that delivered the message. Delivery is direct from each member's read loop — no forwarding goroutines — with the buffer, bound and drop policy of Client.Subscribe / SubscribeQueue applied to the merged output. When fewer members than the success policy requires subscribe, or ctx ends, the members that did are unsubscribed again and the error is returned. Otherwise the token map, keyed by member name, holds the members that subscribed — pass it to UnsubscribeAll, or one entry to Unsubscribe(name, token). The merged output closes once every member's subscription has ended (unsubscribed, refused, or Disconnect).

ch, tokens, err := g.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}})
// tokens["emea"], tokens["apac"]
defer g.UnsubscribeAll(ctx, tokens)

A member whose connection drops reconnects on its own, to its own brokers, while the others carry on; each member's subscriptions come back with it.

Connect / Disconnect / Subscribe run in parallel across members by default. Pass WithGroupSequentialLifecycle if you need deterministic ordering. A group is used once: after Disconnect its methods return ErrClosed.

Use g.Members() or g.Member(name) for direct per-member access (per-member Stats(), a QueuePublisher per member, etc.).

Recipes
Central service over independent clusters

Devices connect to either of two clusters, both always running, that share no data. A cloud service sends them commands, or desired shadow state, without knowing which cluster a device is on. The service keeps a session on both: a ClientGroup with a member per cluster publishes each command to both, and one merged subscription hears the devices' replies from either.

flowchart LR
    service["command service<br/>ClientGroup"] <--> a[("cluster A")]
    service <--> b[("cluster B")]
    a <--> d17["device 17"]
    a <--> d42["device 42"]
    b <--> d63["device 63"]

svc, _ := mqttv5.NewClientGroup(
    []mqttv5.GroupMember{
        {Name: "a", Brokers: []string{"mqtts://a-1.example.com:8883", "mqtts://a-2.example.com:8883"}},
        {Name: "b", Brokers: []string{"mqtts://b-1.example.com:8883", "mqtts://b-2.example.com:8883"}},
    },
    mqttv5.WithGroupSharedOpts(
        mqttv5.WithClientID("command-service"),
        mqttv5.WithCleanStart(false),
        mqttv5.WithSessionExpiry(60*60),
        mqttv5.WithRetryInitialConnect(), // a cluster down at start is tried again
    ),
    mqttv5.WithGroupSuccess(mqttv5.GroupSuccessAny), // one cluster down fails no command
)
// Replies from devices on either cluster, in one channel.
replies, _, _ := svc.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "devices/+/reply", QoS: 1}})
if err := svc.Connect(ctx); err != nil {
    log.Fatal(err)
}

// One command, to both clusters.
expiry := uint32(60)
_, err := svc.Publish(ctx, mqttv5.PublishOptions{
    Topic:                 "devices/42/cmd",
    QoS:                   1,
    Payload:               []byte(`{"desired":{"firmware":"1.4.2"}}`),
    CorrelationData:       []byte("c7"),
    MessageExpiryInterval: &expiry,
})

Acknowledge in the application. MQTT's acknowledgements end at the broker: a QoS 1 PUBACK says cluster A has the command, not that device 42 — perhaps offline, perhaps on cluster B — acted on it. Close the loop as a device shadow or a command RPC does: the device answers each command with a reply carrying its Correlation Data (and, for a shadow, its reported state); the service resends until that reply arrives, devices act once per command ID and answer every copy, and an expiry keeps a copy no device collects from arriving much later. examples/group is the full program: the service's resend loop, the reply matching, and devices on two brokers, one of which comes online late.

sequenceDiagram
    participant S as command service
    participant A as cluster A
    participant B as cluster B
    participant D as device 42 (on A)
    S->>A: PUBLISH devices/42/cmd, correlation c7, expiry 60 s
    S->>B: PUBLISH devices/42/cmd, correlation c7, expiry 60 s
    A->>D: PUBLISH devices/42/cmd
    Note over B: no subscriber: dropped, or held<br/>for a session device 42 left here<br/>until it expires
    D->>A: PUBLISH devices/42/reply, correlation c7, reported state
    A->>S: delivered on the merged subscription
    Note over S: c7 completes the waiting command
At-least-once bridge

Copy messages from one broker to another without losing any, even when the bridge crashes: subscribe on the source with a channel, and ack each message only once the destination has acknowledged its copy.

sequenceDiagram
    participant A as source
    participant Br as bridge
    participant B as destination
    A->>Br: PUBLISH QoS 1, id 41
    Br->>B: PUBLISH QoS 1, id 9 (m.PublishOptions)
    B-->>Br: PUBACK 9
    Br->>A: PUBACK 41 (m.Ack)
    Note over A,Br: a bridge that dies before PUBACK 41<br/>gets message 41 again

src, _ := mqttv5.New(
    mqttv5.WithBroker("mqtts://plant.example.com:8883"),
    mqttv5.WithClientID("bridge-plant-3"),
    mqttv5.WithCleanStart(false),       // resume the source's session after a restart
    mqttv5.WithSessionExpiry(24*60*60), // the source holds messages while the bridge is down
)
dst, _ := mqttv5.New(
    mqttv5.WithBroker("mqtts://cloud.example.com:8883"),
    mqttv5.WithClientID("bridge-plant-3"),
)
// Subscribe before Connect, so the messages held for the bridge reach it.
msgs, _, _ := src.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "plant/#", QoS: 1}})
// Connect dst, then src, then:
for m := range msgs {
    if err := forward(ctx, dst, m); err != nil {
        return err // m stays unacked: the source delivers it again
    }
    _ = m.Ack()
}

// forward publishes m on dst under "site-3/", waiting out dst's
// outages. It returns nil once dst has the message or refused it for
// good, and an error when ctx ends or dst is closed.
func forward(ctx context.Context, dst *mqttv5.Client, m *mqttv5.Message) error {
    opts := m.PublishOptions() // payload and properties as they arrived
    opts.Topic = "site-3/" + m.Topic
    for {
        err := dst.Publish(ctx, opts)
        switch {
        case err == nil:
            return nil
        case errors.Is(err, mqttv5.ErrNotConnected):
            if err = dst.AwaitConnection(ctx); err != nil {
                return err
            }
        case ctx.Err() != nil, errors.Is(err, mqttv5.ErrClosed):
            return err
        default:
            log.Printf("destination refused %s: %v", opts.Topic, err)
            return nil // drop it rather than hold up every message behind it
        }
    }
}

The source keeps the bridge's session, so messages that arrive while the bridge is down wait for it, and any it received but never acked come again when it reconnects or restarts (durable consumers). Never losing one costs the occasional duplicate: after a crash, the source may deliver again a message the destination already has.

This bridge forwards one message per destination round trip. For more, store into a QueuePublisher on queue/file and ack the source once the message is on disk: the queue keeps a window of messages in flight to the destination, retries what can pass and dead-letters what cannot.

out, _ := mqttv5.NewQueuePublisher(dst, q) // q from qfile.Open
for m := range msgs {
    opts := m.PublishOptions()
    opts.Topic = "site-3/" + m.Topic
    opts.QoS = max(opts.QoS, 1) // the queue takes QoS 1 and 2
    if err := out.Publish(ctx, opts); err != nil {
        return err // not stored: the source delivers it again
    }
    _ = m.Ack()
}

To spread the load, run several bridges on a shared subscription ($share/bridge/plant/#): the source hands each message to one of them.

Regional HA

A ClientGroup whose members each fail over within their own region: a broadcast publish reaches every region, and a broker going down moves only its region's member.

flowchart LR
    group["ClientGroup"] --> eu["member eu"]
    group --> us["member us"]
    eu == "connected" ==> eu1[("eu-1")]
    eu -. "standby" .-> eu2[("eu-2")]
    us == "connected" ==> us1[("us-1")]
    us -. "standby" .-> us2[("us-2")]

g, _ := mqttv5.NewClientGroup(
    []mqttv5.GroupMember{
        {Name: "eu", Brokers: []string{"mqtts://eu-1.example.com:8883", "mqtts://eu-2.example.com:8883"}},
        {Name: "us", Brokers: []string{"mqtts://us-1.example.com:8883", "mqtts://us-2.example.com:8883"}},
    },
    mqttv5.WithGroupSharedOpts(mqttv5.WithClientID("pricing")),
    mqttv5.WithGroupSuccess(mqttv5.GroupSuccessAny),
)

With GroupSuccessAny, a publish succeeds while any region takes it; results says which did not.

Quorum writes

Write to independent brokers and succeed once enough of them have the message:

g, _ := mqttv5.NewClientGroup(members, // three independent brokers
    mqttv5.WithGroupSuccess(mqttv5.GroupSuccessQuorum(2)),
)
results, err := g.Publish(ctx, mqttv5.PublishOptions{Topic: "ledger/tx", QoS: 1, Payload: p})
// err is nil once two of the three acknowledged; results has each one's outcome

Publish returns when every member has answered, so ctx bounds how long a slow broker can hold the call.

Sharding with fan-in

Spread topics over an unclustered fleet, each topic always on the same broker, and consume all of them through one subscription:

// Publishers: each topic goes to the same broker while it is connected.
pubs, _ := mqttv5.NewClientGroup(shards,
    mqttv5.WithGroupSharedOpts(mqttv5.WithClientID("orders-pub")),
    mqttv5.WithGroupPublishPolicy(mqttv5.GroupPublishHashByTopic),
)

// Consumers: one merged subscription across every shard; each Ack goes
// back to the broker that delivered the message.
subs, _ := mqttv5.NewClientGroup(shards, mqttv5.WithGroupSharedOpts(mqttv5.WithClientID("orders-sub")))
msgs, tokens, _ := subs.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "orders/#", QoS: 1}})

While a shard's broker is down, its topics pass to the next member. Consumers subscribed to every shard still receive them, but a topic's order is not kept across the switch.

Durable outbox

Publish through outages, broker failovers and process restarts: QueuePublisher on queue/file, the session in store/file, and a broker list.

st, _ := sfile.Open("/var/lib/gateway/session")
q, _ := qfile.Open("/var/lib/gateway/outbound")
cli, _ := mqttv5.New(
    mqttv5.WithBrokers("mqtts://mqtt-1.example.com:8883", "mqtts://mqtt-2.example.com:8883"),
    mqttv5.WithClientID("gateway-12"),
    mqttv5.WithStore(st),
    mqttv5.WithRetryInitialConnect(),
)
pub, _ := mqttv5.NewQueuePublisher(cli, q, mqttv5.WithQueueIdempotencyKey())

pub.Publish returns once the message is on disk, whether a broker is up or not. A restarted process continues the exchanges in flight instead of publishing those messages again. Across a failover between nodes of a cluster the session continues too; between independent brokers the failover is a session loss and in-flight messages go out again, so WithQueueIdempotencyKey adds the mqttv5-msg-id user property for consumers to drop the copy.

Brokers from service discovery
// watchBrokers keeps cli's broker list at the _mqtt._tcp SRV records
// of example.com, checked every minute.
func watchBrokers(ctx context.Context, cli *mqttv5.Client) {
    tick := time.NewTicker(time.Minute)
    defer tick.Stop()
    for {
        _, srvs, err := net.DefaultResolver.LookupSRV(ctx, "mqtt", "tcp", "example.com")
        if err == nil {
            urls := make([]string, 0, len(srvs))
            for _, s := range srvs {
                urls = append(urls, "mqtt://"+net.JoinHostPort(strings.TrimSuffix(s.Target, "."), strconv.Itoa(int(s.Port))))
            }
            err = cli.SetBrokers(urls...)
        }
        if err != nil {
            log.Printf("broker list unchanged: %v", err)
        }
        select {
        case <-ctx.Done():
            return
        case <-tick.C:
        }
    }
}

SetBrokers leaves the connection in use alone; the new list applies from the next connection attempt.

Scaling out consumers

Several instances of a service share the work of one subscription: the broker hands each message to one of them (§4.8.2).

// In every instance:
jobs, _, _ := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "$share/workers/jobs/#", QoS: 1}})

When an instance dies holding a message it never acked, the broker may, at QoS 1, wait for that instance to come back or hand the message to another; at QoS 2 it waits, and the message is lost if the instance's session expires first. Make each instance a durable consumer so it comes back for it.

What combines
Combination Gives
WithBrokers + WithPublisherPool Pool members fail over too, each starting at its own entry of the list; SetBrokers moves them with the main connection.
WithBrokers + WithStore + QueuePublisher The durable outbox.
WithBrokers + WithFollowServerRedirects A 0x9D move replaces the list; a 0x9C redirect takes one attempt.
ClientGroup member with several Brokers Regional HA; a central service whose clusters fail over within themselves.
WithPublisherPool in a GroupMember.Opts A pool per member, over that member's brokers.
NewQueuePublisher(g.Member(name), q) A durable outbox per member.
ClientGroup + $share filters Shared subscriptions on every broker, merged into one output.

Disconnecting

Disconnect(ctx) ends the client gracefully:

  1. it sends DISCONNECT (Normal disconnection) after any acknowledgements already made;
  2. it waits for the connection's goroutines (the read loop runs SubscribeCallback handlers) and closes every subscription's channel or queue;
  3. it waits, bounded by ctx, for session store writes;
  4. with a publish pool, it refuses new publishes with ErrClosed, waits, bounded by ctx, for those in flight on pool members, and disconnects the members.

Disconnect returns ctx's error, joined with the pool members' disconnect errors, if ctx ended before steps 3 or 4 completed. Unfinished QoS 1/2 exchanges of the main connection stay in the session for the next Connect; a pool member's session ends with its connection, which is why step 4 waits for them. Disconnect is idempotent and does not fire OnConnectionDown: the call itself is the "going down" signal. A lifecycle callback may call it; a SubscribeCallback handler must not (see Lifecycle callbacks).

DisconnectWith(ctx, opts) sets the reason code and properties:

expiry := uint32(0)
_ = cli.DisconnectWith(ctx, mqttv5.DisconnectOptions{
    ReasonCode:            mqttv5.ReasonAdministrativeAction,
    ReasonString:          "planned shutdown",
    SessionExpiryInterval: &expiry, // override to drop the session immediately
})

See examples/disconnect.

Options

Client
Broker and transport
Option Default Effect
WithBroker(url) / WithBrokers(urls...) (required) Broker URL(s). mqtt/tcp/mqtts/tls/ssl schemes; default ports filled in. ws/wss via WithDialFunc(ws.DialFunc(opts)). Several URLs fail over: Connect and each reconnect attempt move to the next (Failover).
WithDialFunc(fn) — Replaces the built-in TCP/TLS dial. Takes precedence over WithDialer/WithTLSConfig. Nil rejected.
WithTLSConfig(*tls.Config) — TLS for mqtts://.
WithDialer(*net.Dialer) default Override transport net.Dialer.
Identity and authentication
Option Default Effect
WithClientID(s) broker-assigned MQTT ClientID. Empty = ask the broker to assign one via AssignedClientIdentifier; cli.ClientID() then returns the assigned value after CONNACK.
WithCredentials(user, pass) — CONNECT username + password. Static. For per-attempt rotation use WithConnectPacketBuilder.
WithConnectPacketBuilder(fn) — func(ctx, *ConnectOptions) error. Change the CONNECT's username, password and user properties before each attempt; canonical OAuth-token-rotation hook.
WithAuthenticator(a) — MQTT v5 enhanced auth (CONNECT + re-auth §4.12). Begin(ctx) resolves the credential; client-initiated refresh via Client.Reauthenticate.
Session and CONNECT properties
Option Default Effect
WithCleanStart(b) true, or false when there is session state to resume CleanStart on the initial CONNECT. Unset, the client resumes (CleanStart=0) when it holds unfinished QoS 1/2 flows from an earlier Connect or from WithStore; true always starts fresh and discards that state.
WithCleanStartOnReconnect(b) false CleanStart on every reconnect. False preserves QoS 1/2 session for resume.
WithSessionExpiry(seconds) 300 (5 min) Session Expiry Interval (§3.1.2.11.2). Pass 0 to end the session with the connection. A broker override in CONNACK is honoured.
WithSessionLossPolicy(p) SessionLossRepublish What happens to unacknowledged QoS 1/2 publishes when a connection starts without the session (Session Present = 0): SessionLossRepublish sends them again as new messages in their original order; SessionLossFail completes them with ErrSessionLost.
WithStore(s) none (in memory) session.Store that persists session state across restarts (use store/file). Without it, state survives reconnects only. A failed write of a new message's record fails that Publish; any other failed write stops the client (see Store failures).
WithReceiveMaximum(n) 256 (DefaultReceiveMaximum) QoS 1/2 messages the broker may send unacknowledged, and so how many a subscription whose consumer falls behind holds beyond its buffer. A broker that exceeds it is disconnected with reason 0x93.
WithMaximumPacketSize(n) 0 (no advertised limit) CONNECT property §3.1.2.11.4 — caps the largest packet the broker may send. See note below.
WithInboundTopicAliasMaximum(n) 0 (no inbound aliases) CONNECT property §3.1.2.11.5 — opt into wire compression on inbound PUBLISHes. An alias the broker sends outside 1..n closes the connection (0x94).
WithOutboundTopicAliases() off Replace repeated QoS 0 topics with topic aliases within the broker's Topic Alias Maximum.
WithLenientDecoding() off Tolerate a non-minimal Remaining Length and reserved PINGRESP flags from the broker (logged) instead of disconnecting.
WithRequestResponseInformation(b) false CONNECT property §3.1.2.11.6 — broker returns ResponseInformation in CONNACK.
WithRequestProblemInformation(b) true CONNECT property §3.1.2.11.7 — broker returns ReasonString / UserProperties on errors. Pass false to opt out.
WithConnectUserProperty(k, v) / WithConnectUserProperties(p) — CONNECT user properties; append-style or bulk replace.
WithWill(*WillOptions) — Will message + properties; validated by New.

Session expiry default. 300 s holds the broker session long enough for QoS 1/2 resume across a typical reconnect blip.

Maximum packet size. With the default (no advertised limit), a buggy or hostile broker can send arbitrarily large PUBLISHes. Set it explicitly when broker trust is limited.

Keepalive and timeouts
Option Default Effect
WithKeepAlive(seconds) 30 Requested keep-alive. 0 rejected — use WithoutKeepAlive. A Server Keep Alive in CONNACK replaces it (§3.2.2.3.14); ServerInfo().KeepAlive reports the one in effect.
WithoutKeepAlive() — Request no keep-alive (no PINGREQ unless the broker sets a Server Keep Alive). Rarely correct in production.
WithConnectTimeout(d) 10 s Dial + CONNECT/CONNACK budget.
WithPingTimeout(d) min(10 s, KeepAlive/2) How long to wait for any packet after a PINGREQ before declaring the connection dead. Must be shorter than the keep-alive (New rejects it otherwise).
WithDisconnectFlushTimeout(d) 500 ms Flush budget for the DISCONNECT write on graceful shutdown.
WithReadBufferSize(n) 16 KiB Per-connection read window. Larger windows cut read syscalls for floods of small messages (BenchmarkReceiveWindow).

Keep-alive. A PINGREQ goes out once either direction has been quiet for the keep-alive, so the gap between packets the client sends never exceeds it. After a PINGREQ, any inbound packet counts as the answer (a PINGRESP can sit behind a long inbound backlog). With nothing inbound for PingTimeout the connection is dead: detection takes at most KeepAlive + PingTimeout (40 s with defaults), inside the broker's 1.5×KeepAlive cutoff, so the client drives the reconnect. Time the read loop spends in application code — a SubscribeCallback handler, a SubOnDrop or SubOnDecodeError hook, an Encoding — is not taken for the broker's silence, since the answer may be waiting unread: PINGREQs keep going out and the verdict waits. A broker that really is silent is then detected within two PingTimeouts of the loop being free, or, if no PINGREQ was outstanding, after the usual KeepAlive + PingTimeout.

Reconnect
Option Default Effect
WithReconnectBackoff(b) ExponentialBackoff(1s, 30s, 200ms) Delay before reconnect attempt n. ConstantBackoff(d) also shipped. The attempt count carries over connections that drop sooner than the delay before them.
WithRetryInitialConnect() off A failed first CONNECT is retried in the background like a reconnect and Connect returns nil; Client.AwaitConnection(ctx) waits for the connection. Without it Connect returns the failure: with WithBrokers, once every broker has failed.
WithFollowServerRedirects() off Connect where a 0x9C/0x9D redirect points (see Changing brokers at runtime).
Write path and publishing
Option Default Effect
WithWriteQueueSize(n) 256 Internal MPSC write buffer.
WithWriteBatch(n) 0 (off) Coalesce up to n queued packets per writev syscall. Saves syscalls when concurrent publishers queue packets behind the writer; with one publisher or large payloads there is little to coalesce. The benchmarks do not measure it: measure with your workload before enabling.
WithWriteOverflowPolicy(p) WriteBlock QoS 0 only. See note below.
WithPublishMode(mode) PublishFireAndForget PublishWaitForFlush makes QoS 0 wait until the packet is written; on an idle TCP or Unix connection the caller writes it itself, without copying the payload. ctx bounds the wait; a write it interrupts is finished by the writer goroutine.
WithPublisherPool(N) 0 (off) N publish-only connections to the client's brokers, beside the main one, that Publish spreads over (Publish pool).
WithPublisherPoolRouting(p) PoolRoutingRoundRobin PoolRoutingHashByTopic preserves per-topic ordering.
WithPublisherPoolClientIDFn(fn) "%s-pub-%d"; empty (broker-assigned) when the parent ClientID is Each member's ClientID from the parent's and a 1-based index, e.g. to fit a broker ACL pattern. IDs must be unique across every client of the broker.
WithEncoding(e) none The Encoding of the typed calls (PublishValue, SubscribeAs and their variants), e.g. jsoncodec.Encoding{}; without one they return ErrNoEncoding (Typed publish / subscribe).
WithQoSDowngrade() off Send a publish whose QoS exceeds the broker's Maximum QoS at that maximum instead of failing it with ErrQoSNotSupported. Weakens the guarantee (QoS 2 becomes at-least-once).

Broker limits. Checked before a packet is written; see Broker limits.

Write overflow policy.

  • WriteBlock waits for queue room or ctx.
  • WriteDropNewest returns ErrWriteQueueFull immediately when the writer queue is full. Use it for telemetry, where head-of-line latency on the producer is worse than occasional loss.
  • QoS 1/2 always block on the broker ack regardless.
Subscription defaults
Option Default Effect
WithMaxSubscribeQueueSize(n) DefaultMaxSubscribeQueueSize (65,536) Default SubscribeQueue cap. UnboundedQueue removes it.
WithDropPolicy(p) DropNewest Default drop policy for full sub buffers.
Lifecycle callbacks

These callbacks run one at a time, in event order, on a goroutine the client owns:

  • They may call Disconnect, Connect and SetBrokers.
  • Before each reconnect attempt the client waits for the callbacks so far, so a SetBrokers call from one applies to that attempt, and a slow callback delays it.
  • They run after their event, not inside the call that caused it: OnConnectionUp may run after Connect returns, and a callback for an event before Disconnect may run after Disconnect returns.

Message handlers are different: SubscribeCallback handlers and SubOnDrop hooks run on the connection's read loop, which Disconnect waits for. Calling Disconnect from one would wait for itself; start it on another goroutine (go cli.Disconnect(ctx)). A teardown held up by a handler logs disconnect is waiting for a SubscribeCallback handler after five seconds.

Option Signature / when Effect
WithOnConnectionUp(fn) func(ConnackInfo) — after every successful CONNACK The broker reached (Broker, with WithBrokers the one failed over to) and what it granted: session present, limits, assigned ClientID, effective keep-alive, server reference, ...
WithOnConnectionDown(fn) func() bool — on unexpected disconnect (not user Disconnect) Return false to terminate the supervisor.
WithOnConnectError(fn) per failed CONNECT attempt (dial err, CONNACK refusal, AUTH-loop err) Observability only.
WithOnReconnectAttempt(fn) immediately before each reconnect dial Receives (attempt, brokerURL).
WithOnServerDisconnect(fn) func(DisconnectInfo) — on broker-initiated DISCONNECT Reason code, reason string, server reference, user properties.
WithOnServerRedirect(fn) func(ServerRedirect) — CONNACK or DISCONNECT with 0x9C/0x9D and a Server Reference Observability, or custom handling instead of WithFollowServerRedirects.
WithOnResubscribeError(fn) func(SubscriptionToken, error) — the broker refused filters when the client re-subscribed after a session loss The token's Err() is set when no filter is left.
WithOnReauthenticated(fn) when a re-authentication started with Reauthenticate concludes with broker AUTH 0x00 Observability only.
WithOnStoreFailure(fn) func(error) — once, after a WithStore write failed and the client stopped because of it The error is a *StoreError (errors.Is(err, ErrStoreFailed)). Call Connect to start again from what the store holds.
Observability
Option Default Effect
WithLogger(*slog.Logger) slog.Default() Structured logging.
WithStats() — Enable in-memory counters for Client.Stats() (off by default to keep the hot path branch-predictor friendly).
Per-subscribe

SubscribeOption applies to Subscribe, SubscribeQueue, SubscribeCallback, plus their typed (SubscribeAs[T], …) and ClientGroup variants.

Option Effect
SubBuffer(n) Channel buffer size (Subscribe only): the QoS 0 messages held before one is dropped; QoS 1/2 have room for Receive Maximum more. Default DefaultSubscribeBuffer (64).
SubMaxQueueSize(n) Queue cap (SubscribeQueue only); under DropNewest, QoS 1/2 have room for Receive Maximum more. 0 keeps the client default; UnboundedQueue removes the cap.
SubDropPolicy(p) DropNewest / DropOldest. SubscribeQueue honours both, and DropOldest evicts whatever the QoS; chan-based Subscribe returns ErrChanDropOldestUnsupported when DropOldest is set explicitly.
SubOnDrop(fn) Hook run on the read goroutine before a dropped message is acked; every field is readable. Must not block or call Disconnect.
SubAutoAck() Opt-in: the dispatcher acks each delivery before handing it to the consumer, so a full buffer or queue drops QoS 1/2 messages too. See note below. Ignored by SubscribeCallback.
SubEncoding(e) Decode this typed subscription with e instead of the client's Encoding. Ignored by untyped subscriptions.
SubOnDecodeError(fn) func(*Message, error), run on the read goroutine for each payload a typed subscription's Encoding rejects, before the message is acked. Must not block or call Disconnect.
SubZeroCopy() Deliver messages whose Topic / Payload / Properties alias the network frame (valid until every receiving subscription has acked; for callbacks, until return). Saves the copy: one allocation per message at QoS 0. Also accepted by SubscribeCallback.

SubAutoAck trade-offs.

  • m.Ack() is a no-op; the message is an owned copy, safe to retain.
  • Breaks at-least-once semantics: a consumer crash between delivery and processing has nothing to replay.
  • Reach for it on QoS 0 / observational consumers.
Per-QueuePublisher
Option Type Default Effect
WithQueueWindow(n) int 32 (DefaultQueueWindow) Messages in flight at once; the broker's Receive Maximum still applies. Must be ≥ 1.
WithQueueMaxSize(n) int 0 (unbounded) Bound on queued messages, enforced atomically by the queue.
WithQueueDropPolicy(p) DropPolicy DropNewest At the bound: DropNewest returns ErrQueueFull; DropOldest evicts the oldest messages not in flight and dead-letters them.
WithQueueTTL(d) time.Duration 0 (none) Longest a message may wait, from Publish; the remaining lifetime is sent as Message Expiry Interval.
WithDeadLetter(fn) func(QueueEntry, error) none Messages removed without the broker accepting them, with the reason. Called from internal goroutines, possibly concurrently; must not block.
WithQueueClassifier(fn) func(error) bool PermanentPublishError Which failures are dead-lettered rather than retried.
WithQueueRetryBackoff(b) Backoff DefaultQueueRetryBackoff (1 s doubling to 1 min, ±0.5 s) Delay before retrying a message whose publish failed for a reason that may pass.
WithQueueIdempotencyKey() — off Adds the mqttv5-msg-id user property with the message's ID.
Per-ClientGroup
Option Type Default Effect
WithGroupSharedOpts(opts...) ...Option none Client options applied to every member before its GroupMember.Opts.
WithGroupPublishPolicy(p) GroupPublishPolicy GroupPublishBroadcast Broadcast, RoundRobin or HashByTopic (see Parallel sessions).
WithGroupSuccess(s) GroupSuccess GroupSuccessAll How many members must succeed for a broadcast Publish or a Subscribe; GroupSuccessAny, GroupSuccessQuorum(n).
WithGroupSequentialLifecycle() — parallel Connect, Disconnect and Subscribe members one after another.
GroupMember field Default Effect
Brokers (required) The member's broker URLs: one, or several it fails over between as WithBrokers does. WithBroker/WithBrokers in Opts or WithGroupSharedOpts is rejected.
Name member-N (1-based) Key in the Subscribe token map and for ClientGroup.Member; must be unique.
Opts none Client options for this member, applied after WithGroupSharedOpts.

Errors

Branch on sentinel errors with errors.Is. Typed errors (*ReasonCodeError, *SubscribeError, …) carry the details; reach them with errors.As.

Connection lifecycle
Error Source Meaning
ErrNotConnected Publish, Reauthenticate No live connection; from Publish it also guarantees the message was not sent. Retry / wait for reconnect (AwaitConnection).
ErrAlreadyConnected Connect Connect called twice.
ErrClosed any after Disconnect Client torn down.
ErrConnectRefused Connect Broker non-success CONNACK reason; the error is a *ReasonCodeError carrying the code, reason string and server reference.
joined errors Connect with WithBrokers No broker accepted: one error per broker, each prefixed with its URL; errors.Is / errors.As see every one.
*ProtocolError Connect, connection drops The broker broke MQTT v5 (e.g. Session Present for CleanStart=1, a packet above the client's Maximum Packet Size, more in-flight messages than the client's Receive Maximum). The client sent DISCONNECT with Reason and reconnects; counted in Stats().ProtocolErrors.
ErrInvalidSessionExpiry DisconnectWith Non-zero Session Expiry when CONNECT sent 0 MQTT-3.14.2-2; nothing is sent.
ErrUnexpectedPacket various Broker sent an unexpected packet for the current state. Treat as protocol bug.
ErrMissingBroker New No URLs supplied.
ErrInvalidBrokerURL New, SetBrokers URL failed to parse or has unsupported scheme. WithDialFunc relaxes scheme validation.
Re-authentication
Error Source Meaning
ErrNoAuthenticator Reauthenticate Called without WithAuthenticator.
ErrReauthInProgress Reauthenticate A re-auth is already in flight on this connection (calls are single-flighted).
ErrReauthRejected Reauthenticate Broker rejected re-auth via DISCONNECT; wrapped with the reason code. Connection torn down; supervisor reconnects.
Subscribe
Error Source Meaning
*UnsubscribeError Unsubscribe The broker refused to remove some filters (UNSUBACK ≥ 0x80); the local subscription is closed anyway.
ErrChanDropOldestUnsupported Subscribe (chan) Explicit SubDropPolicy(DropOldest) on the channel-based Subscribe. Use SubscribeQueue for DropOldest.
ErrNilHandler SubscribeCallback Handler argument was nil.
ErrSharedSubsUnsupported Subscribe* Broker disabled $share/... in CONNACK.
ErrWildcardSubsUnsupported Subscribe* Broker disabled + / # in CONNACK.
*SubscribeError Subscribe* The broker refused some filters; Results has each filter's outcome. With some granted the subscription is active for those; with none its channel/queue is closed. errors.Is matches the refusal's sentinel (e.g. ErrNotAuthorized).
*RejectedByDisconnectError Subscribe*, Unsubscribe Instead of answering, the broker closed the connection with a DISCONNECT blaming the packet (0x81 Malformed Packet, 0x82 Protocol Error, …) each of the 3 times it was sent — e.g. mosquitto for a filter deeper than 200 levels. The client gives up on it and stays connected; the subscription is closed.
Publish
Error Source Meaning
ErrWriteQueueFull Publish (QoS 0) Writer queue at capacity AND client configured with WithWriteOverflowPolicy(WriteDropNewest). The publish never reached the wire.
*ReasonCodeError Publish (QoS 1/2) The broker refused the message (PUBACK/PUBREC reason ≥ 0x80). errors.Is matches ErrNotAuthorized, ErrQuotaExceeded, ErrTopicNameInvalid, ErrPayloadFormatInvalid, ErrPacketTooLarge; errors.As gives the code and reason string.
ErrQoSNotSupported Publish QoS above the broker's Maximum QoS (see WithQoSDowngrade).
ErrInvalidTopic Publish, Subscribe*, Unsubscribe Topic name with a wildcard, NUL or invalid UTF-8; malformed filter or shared subscription.
ErrInvalidField Publish, Subscribe*, Connect Another option MQTT v5 forbids (DUP on QoS 0, No Local on a shared subscription, Retain Handling 3, ...).
ErrFieldTooLong any A string or binary field over 65,535 bytes.
ErrTopicAliasInvalid Publish Topic Alias above the broker's maximum, or a QoS 1/2 publish with an alias but no topic.
ErrRetainNotSupported Publish Retained message on a broker with Retain Available = 0.
ErrPacketTooLarge Publish, Subscribe*, Unsubscribe Packet above the broker's Maximum Packet Size; nothing was sent.
ErrNoEncoding PublishValue, SubscribeAs and their variants No Encoding: set WithEncoding, or SubEncoding for one subscription. Nothing was sent.
ErrSessionLost Publish (QoS 1/2) Session lost with WithSessionLossPolicy(SessionLossFail).
ErrMessageExpired Publish (QoS 1/2) Message Expiry ran out before a republish after session loss.
ErrPacketIDsExhausted Publish, Subscribe* All 65,535 packet identifiers in use until ctx ended.
QueuePublisher
Error Source Meaning
ErrQueueClosed QueuePublisher.Publish, PublisherQueue methods After Close.
ErrQueueFull QueuePublisher.Publish; dead letters At WithQueueMaxSize under DropNewest, or under DropOldest when every queued message is in flight. Dead letters carry it for evicted messages.
ErrQoS0NotQueueable QueuePublisher.Publish QoS 0 has no acknowledgement to queue for.
ErrMessageExpired dead letters The message's TTL or Message Expiry ran out before it was sent.
*ReasonCodeError dead letters The broker refused the message with a code WithQueueClassifier calls permanent.
ClientGroup
Error Source Meaning
*GroupError ClientGroup.Publish, ClientGroup.Subscribe Fewer members succeeded than WithGroupSuccess requires; Results has each member's outcome and errors.Is sees their errors.
Transport
Error Source Meaning
transport.ErrUnknownScheme transport.Dial The URL's scheme is not one the built-in dialer knows. New rejects such a broker URL up front with ErrInvalidBrokerURL.
transport.ErrMissingHost transport.Dial The URL has no host.

Reliability semantics

Connection

Connect. Blocks until the CONNACK, or until ctx ends. With WithBrokers it tries each broker once, from a random one on. Without WithRetryInitialConnect() it returns the failure, every broker's when there are several. With it, Connect returns nil, the supervisor retries with the reconnect backoff, and Client.AwaitConnection(ctx) waits for the connection; until then Publish returns ErrNotConnected, Subscribe waits for the connection, and a QueuePublisher keeps queuing.

Reconnect. ExponentialBackoff(1s, 30s, 200ms) by default. With WithBrokers, URLs rotate per attempt and a successful connect sticks. The backoff starts over only after a connection outlives the delay that preceded it, so a broker that accepts the CONNECT and drops the connection at once is retried at a growing interval, not in a tight loop.

Keep-alive. After a PINGREQ, any packet from the broker counts as the answer. With none within PingTimeout the connection is treated as dead and the supervisor redials — unless application code held the read loop meanwhile, which only delays the verdict (see the keep-alive note).

Server-initiated DISCONNECT. WithOnServerDisconnect(fn) receives a DisconnectInfo (reason, reason string, server reference, user properties) before OnConnectionDown. Redirects (0x9C/0x9D) are also reported to WithOnServerRedirect and followed with WithFollowServerRedirects.

Sessions

Session resume (Session Present = 1). Unacknowledged QoS 1/2 PUBLISHes are resent in their original send order with DUP=1 (§4.4, §4.6); a QoS 2 message that already got its PUBREC resends PUBREL, never the PUBLISH; PUBRELs follow PUBREC order. Acks the application made while disconnected are sent on resume. The caller stays blocked on Publish across the drop.

Session loss (Session Present = 0, or CleanStart = 1). Inbound state is discarded, so the new session's messages are never mistaken for duplicates, and an Ack on a message from the old session is a no-op. Unacknowledged outbound QoS 1/2 messages follow WithSessionLossPolicy: republished as new messages (the default; a duplicate is possible, and QoS 2 cannot stay exactly-once across the loss) or failed with ErrSessionLost. A broker that reports a present session for a CleanStart=1 CONNECT is a protocol error (DISCONNECT 0x82). With WithBrokers, failover to a broker that does not share the session (not a node of the same cluster) is a session loss.

Send quota. At most the broker's Receive Maximum QoS 1/2 PUBLISHes are in flight (§4.9); further publishes wait in order.

Packet identifiers. Owned per flow (publish, subscribe, unsubscribe) and released only when that flow completes, so a stray ack cannot free an identifier another flow is using.

Duplicates from the broker. A resent PUBLISH (DUP) for a message the application still holds is neither delivered again nor acknowledged on the application's behalf; a QoS 2 duplicate after PUBREC gets PUBREC again; PUBREL for an unknown identifier gets PUBCOMP 0x92. Acks for unknown or mismatched packet identifiers are ignored and counted in Stats().AcksIgnored.

Delivery

Acknowledgement order. A QoS 1 PUBACK is held until m.Ack() and flushed in §4.6 arrival order. A QoS 2 PUBREC is held until m.Ack(); PUBCOMP goes out automatically when PUBREL arrives.

A consumer that falls behind. Holding the acknowledgement until m.Ack() is also the flow control: the broker sends at most Receive Maximum QoS 1/2 messages the client has not acknowledged, so a channel or queue subscription keeps room for that many beyond its buffer and drops none of them. QoS 0 messages beyond the buffer are dropped and acked (see Subscribing).

One message, several subscriptions. A PUBLISH matching several subscriptions gives each its own *Message handle over the same bytes. The broker's ack (PUBACK or PUBREC) goes out once every handle has been acked. m.Ack() is idempotent per handle: a second call is a no-op and never affects another message or handler.

Topic and payload lifetime. Owned by the message and valid for as long as you hold it. With SubZeroCopy() they alias the network frame and are valid only until every receiving subscription has acked (callbacks: until return); use m.CloneTopic() / m.ClonePayload() to keep them longer.

Subscriptions

Across a drop. A SUBSCRIBE or UNSUBSCRIBE still waiting for its answer when the connection drops is sent again on the next one, in the order the calls were made; Subscribe and Unsubscribe keep waiting until the answer arrives or ctx ends. A Subscribe made while the client reconnects waits for the next connection the same way.

Before Connect. A Subscribe before Connect (or between Disconnect and the next Connect) returns at once, and its SUBSCRIBE goes out first on the connection Connect makes. Its route is in place before the CONNACK, so the messages a resumed session holds — which the broker sends right after the CONNACK — reach it; token.Results() reports the SUBACK once it arrives. A PUBLISH no subscription matches is acknowledged and dropped, counted in Stats().InboundUnrouted; the first on each connection is logged at Warn unless it was for a subscription being unsubscribed, since on a resumed session it is most likely a queued message that arrived before the application subscribed. A filter that connection's CONNACK rules out (shared or wildcard subscriptions unavailable) ends the subscription with that error instead of being sent. If Connect fails, its subscriptions end with it. When the broker has no session, pending UNSUBSCRIBEs complete without being sent and every subscription is subscribed again, oldest first. Filters the broker refuses then end the subscription's share of them: WithOnResubscribeError is called, and a subscription left with no filters closes with token.Err() set. A SUBSCRIBE or UNSUBSCRIBE the broker answers by disconnecting (DISCONNECT blaming a packet it received) is sent at most three times, then fails with *RejectedByDisconnectError. A filter the broker never granted is not unsubscribed.

Shared filters. A broker keeps one subscription per exact filter, and a SUBSCRIBE for a filter it already holds replaces it (§3.8.4). Subscriptions of one client with the same filter therefore share the broker's: each receives every message for it; the options of the most recent Subscribe (QoS, No Local, Retain As Published, Retain Handling) apply to all of them until the last one unsubscribes; retained messages sent for a later Subscribe reach the earlier ones as well. Unsubscribe sends an UNSUBSCRIBE only for filters no other subscription of the client uses.

Subscription Identifiers. When the broker supports them (CONNACK), every SUBSCRIBE carries a Subscription Identifier the client allocates, and a PUBLISH tagged with identifiers reaches only the subscriptions it was sent for. With overlapping filters (a/# and a/b) a broker may send one copy per matching subscription; each subscription then receives exactly one. Without identifiers every matching subscription receives every copy. A PUBLISH tagged only with identifiers this client never gave a subscription to its topic was sent for a subscription of an earlier process, whose session this one resumed: it is matched by topic, as without identifiers. One tagged with an identifier the client has replaced, or is unsubscribing, is dropped.

Broker capabilities. When the CONNACK disables shared subscriptions, wildcards or Subscription Identifiers, Subscribe* fails before anything is sent.

Teardown. Unsubscribe closes the subscription's channel or queue before it returns, then reports the broker's answer; Disconnect closes all of them. Each closes exactly once; a PUBLISH already being dispatched to a closed subscription is acked instead of delivered. Disconnect clears all routes, so Connect followed by Subscribe starts clean.

Protocol

Validation. Every packet from the broker is checked against MQTT v5 before it is acted on (fixed-header flags, minimal Remaining Length, which properties each packet may carry and their values, UTF-8 strings, topic names, packet identifiers, reason codes, topic aliases). A violation closes the connection with DISCONNECT and the spec's reason code (0x81 Malformed Packet, 0x82 Protocol Error, or a specific one such as 0x93, 0x94, 0x95), then the supervisor reconnects; each is logged at Error and counted in Stats().ProtocolErrors. WithLenientDecoding() tolerates the two violations real brokers produce harmlessly (a non-minimal Remaining Length, reserved PINGRESP flag bits). Outbound packets are validated the same way: an invalid topic, filter or option fails the call with ErrInvalidTopic, ErrInvalidField or ErrFieldTooLong before anything is sent.

Topic aliases. Outbound aliases are opt-in (WithOutboundTopicAliases()): QoS 0 publishes then replace a repeated topic with an alias while the broker's Topic Alias Maximum lasts; the alias is allocated, encoded and queued under one lock per connection, so a use never overtakes its registration. QoS 1/2 always carry the full topic, since they may be resent on a connection that knows no aliases. Inbound aliases need WithInboundTopicAliasMaximum(n); an alias of 0 or above n is a DISCONNECT 0x94, an alias used before it was registered a 0x82.

Re-authentication (§4.12). Client-initiated with Reauthenticate(ctx): AUTH 0x19, then Begin/Continue, until the broker's 0x00 Success. The client always replies 0x18 Continue; only the server concludes, and an inbound 0x00 is terminal. A rejection returns ErrReauthRejected and the client reconnects with a fresh CONNECT.

The exchange is validated: during CONNECT the broker may only send AUTH 0x18 under the CONNECT's Authentication Method, and a CONNACK may not name another method; after CONNACK a broker AUTH is only valid inside a re-authentication the client started (a broker cannot initiate one), again under the same method. Anything else is a protocol error (DISCONNECT 0x82). Authenticators that do I/O to answer a challenge can implement ContextAuthenticator to be cancelled at the connect timeout.

Performance

Measured with the benchmark suite: mqttv5 against eclipse/paho.golang (autopaho and its packets codec) and eclipse/paho.mqtt.golang (paho3, MQTT 3.1.1), every client running the same scenario code through the same pinned mosquitto broker. Each table is generated by benchmarks/cmd/benchtab from a recording made with benchmarks/scripts/run.sh, which keeps the commit, toolchain, host, broker and load average it ran under; the recordings stay out of the repository, and each table names the date and commit of its own.

This page shows allocations, which vary far less between hosts and runs than timings do; they count every goroutine in the benchmark process, and pools, batching and timers still move them a little. Timings — call latency, throughput, CPU per message, open-loop latency percentiles, reconnect time — with the conditions they were measured under and how to read them, are in benchmarks/README.md.

Codec

Allocations to decode one PUBLISH from a stream (props=five: a content type and five user properties), then read its properties:

props size eclipse mqttv5
none 64B 22 0
none 256B 22 0
none 1KiB 24 0
none 16KiB 32 0
five 64B 93 0
five 256B 93 0
five 1KiB 95 0
five 16KiB 103 0

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

Allocations to encode one PUBLISH and write it, with each of mqttv5's encoders: mqttv5-header writes the caller's payload as it is (a QoS 0 publish its caller writes on a TCP connection), mqttv5-pooled copies the packet into a pooled buffer (a QoS 0 publish queued for the writer goroutine), and mqttv5-owned copies it into a buffer of its own that the session keeps for retransmission (every QoS 1/2 publish). See the codec section:

props size eclipse mqttv5-header mqttv5-pooled mqttv5-owned
none 64B 9 0 0 1
none 256B 9 0 0 1
none 1KiB 9 0 0 1
none 16KiB 9 0 0 1
five 64B 12 0 0 1
five 256B 12 0 0 1
five 1KiB 12 0 0 1
five 16KiB 12 0 0 1

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

Publishing

Allocations per Publish call in the publishing client. QoS 0 returns once the packet is written (mqttv5 with PublishWaitForFlush, which on this TCP connection writes it on the caller's goroutine without copying the payload); QoS 1 and 2 return when the broker has acknowledged. The packet a QoS 1/2 publish keeps for retransmission is one exact-size copy.

qos size mqttv5 autopaho paho3
0 64B 0 15 23
0 1KiB 0 15 23
0 1MiB 0 15 23
1 64B 5 53 42
1 1KiB 5 53 42
1 1MiB 7 55 ±2% 43
2 64B 6 92 62
2 1KiB 6 92 62
2 1MiB 8 93 63

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

qos size mqttv5 autopaho paho3
0 64B 0B 600B 1.305KiB
0 1KiB 0B 600B 2.305KiB
0 1MiB 0B 600B 1.009MiB
1 64B 424B 4.309KiB 1.641KiB
1 1KiB 1.447KiB 5.309KiB 2.642KiB
1 1MiB 1.009MiB 1.012MiB 1.010MiB
2 64B 440B 7.191KiB 2.145KiB
2 1KiB 1.462KiB 8.191KiB 3.146KiB
2 1MiB 1.009MiB 1.015MiB 1.010MiB

B/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

Receiving

Allocations per delivered message, counted across the benchmark process; the raw publisher in it allocates nothing per message, so they are the subscriber's. callback is a handler the library calls; chan is mqttv5's channel and, for autopaho, the channel an application forwards into from its handler. A QoS 1 message is acknowledged when the callback returns, or, in chan and queue mode, when a consumer takes it.

mode qos size mqttv5 autopaho paho3
callback 0 64B 2 27 16
callback 0 1KiB 2 29 16
callback 1 64B 5 40 33
callback 1 1KiB 5 42 34
chan 1 64B 5 40 —
chan 1 1KiB 5 42 —
queue 1 64B 6 — —
queue 1 1KiB 6 — —

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

Publish-to-delivery round trip, both clients in one process:

qos size mqttv5 autopaho paho3
0 64B 5 45 42
0 1KiB 5 47 42
0 1MiB 13 66 42
1 64B 13 95 78
1 1KiB 13 97 78
1 1MiB 22 ±5% 117 ±1% 80
2 64B 15 173 120
2 1KiB 15 175 120
2 1MiB 25 ±4% 194 121 ±1%

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

mqttv5's receive path by delivery mode, measured in-process without a broker (BenchmarkReceive, 64-byte payloads):

qos consumer owned zerocopy
1 callback 4 3
1 chan 3 2
1 queue 4 3
0 callback 2 1

allocs/op: median of 10 runs ±95% CI; recorded 2026-10-09 at 80beb93780a2.

WithStats is off in these runs; its counters are atomic adds behind a nil check.

Operations

Health and metrics

Client.Connected() reports a live connection, and Client.ServerInfo() which broker it reached (Broker) and what that broker granted. The lifecycle callbacks report connection events as they happen.

Client.Stats() returns a snapshot of in-memory counters. Opt in via WithStats() — when off, the hot path skips every atomic increment and Stats() returns the zero value.

cli, _ := mqttv5.New(
    mqttv5.WithBroker(broker),
    mqttv5.WithStats(),
)
// ...
s := cli.Stats()
fmt.Printf("sent=%d acked=%d inflight=%d connects=%d failures=%d\n",
    s.PublishesSent, s.PublishesAcked, s.PublishesInflight,
    s.Connects, s.ConnectFailures)
Field Kind What it measures
Connects counter Successful CONNECT/CONNACK handshakes
ConnectFailures counter Connection attempts that failed, for any reason (dial, TLS, CONNACK refusal, authentication)
Disconnects counter Connections lost, for any reason but Disconnect
ServerDisconnects counter DISCONNECT packets the broker sent
PingTimeouts counter Connections declared dead after a PINGREQ went unanswered
PublishesSent counter Publish calls that returned nil — QoS 0 once queued (or written, with PublishWaitForFlush), QoS 1/2 once acknowledged — and QueuePublisher messages the broker accepted
PublishesAcked counter PUBACK and PUBCOMP packets received, refusals included, and those for publishes whose caller stopped waiting
PublishesReplayed counter QoS 1/2 publishes resent after a reconnect
PublishesInflight gauge QoS 1/2 publishes waiting for their acknowledgement
SendQuota gauge QoS 1/2 publishes the broker's Receive Maximum still allows in flight
InboundPublishes counter PUBLISH packets received
InboundDropped counter Received messages dropped because a subscription's buffer or queue was full
InboundUnrouted counter Received messages no subscription matched: acked and dropped
DecodeFailures counter Payloads a typed subscription's Encoding rejected: acked and not delivered
SubscribesSent, UnsubscribesSent counter SUBSCRIBE and UNSUBSCRIBE packets sent
SubscriptionsActive gauge Subscriptions registered on the client
PoolFallbacks counter Publishes no pool member could send, handed to the main connection, whether or not it could send them
PoolConnected gauge Publish pool members with a live connection
AcksIgnored counter Acknowledgements for an unknown or mismatched packet identifier
StoreErrors counter Failed session.Store operations
ProtocolErrors counter Broker protocol violations that closed the connection

Counters start at zero in New and are never reset: they span every Connect and Disconnect of the client. With a publish pool, every field covers the members' connections as well as the main one.

The library has no metrics dependency. Read Stats() once per collection — from a Prometheus Collector, an OpenTelemetry observable instrument's callback, or an expvar.Func — and export counters as counters and gauges as gauges; reading it once per field takes the gauges' locks once per field. examples/stats prints a snapshot every second.

Worth alerting on:

  • ConnectFailures or Disconnects rising: the connection is flapping (Troubleshooting).
  • InboundDropped rising: a consumer is behind the stream.
  • InboundUnrouted rising outside unsubscribes: messages reach no subscription, typically a resumed session's queued messages arriving before the application subscribed.
  • DecodeFailures rising: a publisher sends payloads the subscription's Encoding cannot decode.
  • PoolConnected below the pool size, or PoolFallbacks rising: pool members cannot connect.
  • StoreErrors or ProtocolErrors above zero.
  • PublishesInflight at the broker's Receive Maximum (SendQuota 0) for long: the broker is not acknowledging.
Logs

Structured log/slog through WithLogger (default slog.Default()). Normal message traffic logs nothing.

  • Info: connects and disconnects.
  • Warn: recoverable problems: reconnects, refused filters, stray acknowledgements, typed decode failures, a teardown waiting on a handler.
  • Error: protocol violations and store failures.
Troubleshooting
Symptom Check Then
Connect returns ErrConnectRefused errors.As(err, &rce) for the CONNACK reason code 0x86/0x87: credentials or authorization; 0x9C/0x9D: the broker redirects, see WithFollowServerRedirects
Repeated reconnects Stats().Connects, ProtocolErrors; Error logs Protocol errors name the violation; a broker that sends harmless non-minimal lengths needs WithLenientDecoding()
Publish blocks Stats().PublishesInflight at the broker's Receive Maximum The broker is not acknowledging; the call returns when it does or ctx ends
QoS 1/2 Publish slow while subscriptions are busy how long SubscribeCallback handlers and SubOnDrop hooks take They run on the read loop that also reads this client's publish acknowledgements: move slow work to a channel or queue subscription, or publish through a publish pool
ErrWriteQueueFull WithWriteOverflowPolicy(WriteDropNewest) is set and the writer is behind Raise WithWriteQueueSize, or use the default blocking policy
Typed messages missing; Warn log payload decode failed Stats().DecodeFailures; the topic in the log; SubOnDecodeError The publisher sends another format or shape than the subscription's Encoding and type expect: fix the publisher, or use SubEncoding / another type for that topic
A durable consumer misses the messages queued while it was down Warn log a message matched no subscription; Stats().InboundUnrouted It subscribes after Connect, and the broker sends the queued messages right after the CONNACK: subscribe before Connect (Durable consumers)
Messages missing on a subscription Stats().InboundDropped; SubOnDrop The consumer is slower than the stream. Only QoS 0 messages are dropped, unless SubAutoAck or DropOldest is set: raise SubBuffer / SubMaxQueueSize, subscribe at QoS 1, consume faster, or use a callback (backpressure)
Duplicate messages after a restart Session Present in OnConnectionUp; the store Without a durable WithStore, a restart loses QoS 1/2 state; WithSessionLossPolicy decides what happens to unacknowledged publishes
ErrLocked from store/file / queue/file another process holds the file One process per store directory
Client stopped, WithOnStoreFailure fired, Publish returns ErrStoreFailed the *StoreError's Op and Err; the Error log; free space and permissions on the store directory Fix the store, then Connect again: it reloads the session from the store (see Store failures)
Memory grows with large messages the broker's message sizes Set WithMaximumPacketSize to the largest message you expect
A publish pool is no faster than one connection Stats().PoolConnected against the pool size; PoolFallbacks Members not connected: check the broker's connection limit and that its ACLs admit <ClientID>-pub-<n>; members log with component=publisher-pool. All connected: the round trip is not what limits publishing (Publish pool)
Own messages arrive on a No Local subscription WithPublisherPool is set Pooled publishes come from the members' ClientIDs; skip your own messages in the handler
Disconnect does not return; Warn log disconnect is waiting for a SubscribeCallback handler goroutine dump (SIGQUIT) for the handler's stack A handler or SubOnDrop hook is blocked, or called Disconnect itself: make it return, and run Disconnect from another goroutine
Reconnects start late a lifecycle callback that takes long to return The client waits for the callbacks before each attempt: keep them short, hand slow work to another goroutine
Store failures

Every packet that depends on a WithStore record waits until the record is written: a PUBLISH for its record, a PUBREL for the AwaitPubcomp phase, a PUBREC for the inbound AwaitPubrel record and a PUBCOMP for that record's deletion. One flow's writes reach the store in the order the client decided them, also across a reconnect.

A failed write of a new message's record fails that Publish with a *StoreError. Any other failed write means the store no longer holds what the session has done, so the client stops instead of carrying on without the guarantee the store is there for, much as a crash would stop it:

  1. it drops the session state it holds in memory and sends nothing more that depends on it;
  2. it sends DISCONNECT 0x80 (the broker publishes the Will) and ends as Disconnect would;
  3. QoS 1/2 Publish calls still waiting return the *StoreError (errors.Is(err, ErrStoreFailed)); their messages stay in the store;
  4. WithOnStoreFailure fires;
  5. the next Connect reloads the session from the store and continues every exchange from the phase it holds.

The client treats every store error as final. A session.Store over a backend with transient failures (a network database, say) retries them itself within the ctx it is given. store/file returns an error only when the file system does: check free space and the Error log, then Connect again.

Security

  • TLS. mqtts://, ssl://, tls:// and wss:// verify the broker against the system roots by default; WithTLSConfig supplies CAs, client certificates or a server name. Never set InsecureSkipVerify outside tests.
  • Credentials. WithCredentials and WithConnectPacketBuilder (per-attempt rotation) keep secrets in memory only; the client never logs them. Enhanced authentication (WithAuthenticator) supports SCRAM-style challenge/response and re-authentication (Reauthenticate).
  • Untrusted input. Every packet from the broker is validated before it is acted on, and a violation closes the connection (see Reliability semantics). Memory a broker can make the client hold is bounded: per packet by WithMaximumPacketSize (unset: the decoder holds at most eight times the bytes actually received until a packet is complete), per subscription by SubBuffer / SubMaxQueueSize for QoS 0 messages and by WithReceiveMaximum for QoS 1/2 ones.
  • Persistence. store/file and queue/file write message payloads to disk unencrypted, with 0600 permissions by default under a 0700 directory; encrypt the volume if payloads are sensitive.
  • Reporting. Report vulnerabilities privately through GitHub security advisories on this repository.

Architecture

Outbound, a publish takes the shortest path the connection allows:

flowchart LR
    pub["Publish"] -- "QoS 1/2: packet ID,<br/>send quota, record" --> engine["session engine"] --> store[("session.Store")]
    pub -- "idle TCP or Unix socket" --> direct["caller's goroutine,<br/>direct write"] --> sock(("socket"))
    pub -- "busy writer, TLS, WebSocket,<br/>fire-and-forget QoS 0" --> writer["writer goroutine<br/>queue, writev batches"] --> sock

Inbound, acknowledgements leave in arrival order, from the read loop itself when the socket takes them:

flowchart LR
    sock(("socket")) --> reader["read loop<br/>decode, validate,<br/>trie match"] --> subs["channel, queue<br/>or callback"] -- "Ack" --> engine["session engine<br/>acks in arrival order"]
    engine -- "ready when the read loop<br/>waits: non-blocking write" --> out(("socket"))
    engine -- "the rest" --> writer["writer goroutine"] --> out

One goroutine per connection drives read → decode → trie match → handlers (sync). On a TCP or Unix connection, a publish that waits for its write (QoS 1/2, or QoS 0 with PublishWaitForFlush) writes on its own goroutine when nothing is queued: a QoS 0 header and its payload in one writev, without copying the payload; a QoS 1/2 publish the packet the session keeps for retransmission. If the caller's ctx ends mid-write, a deadline stops the write and the writer goroutine sends the rest before anything else. Otherwise — TLS and WebSocket connections, a busy writer, and fire-and-forget QoS 0 — one writer goroutine per connection drains an MPSC channel of encoded packets to the socket, coalescing queued packets into one writev when WithWriteBatch is set; a write lock keeps the two paths from interleaving, and a publish never overtakes a packet its caller queued earlier. Before the read loop waits for more input, it writes the acknowledgements and PUBRELs it has made ready itself — on an idle TCP or Unix connection, on unix systems — as far as the socket takes them without waiting; the writer goroutine sends the rest, so a broker that stops reading never stops the client reading. A supervisor goroutine owns the connection lifecycle: reconnect with backoff, session resumption, and re-issuing subscriptions.

Frames. The decoder reads through a WithReadBufferSize window (16 KiB). A body up to 64 KiB goes into a pooled frame from one of five size classes (256 B – 64 KiB). A larger body is read into buffers that grow as its bytes arrive — never more than eight times what the broker has actually sent — and the final buffer is never pooled. A declared length therefore costs nothing until its bytes arrive; WithMaximumPacketSize rejects oversized packets before reading them.

Messages. A delivered message copies topic, payload and properties out of a pooled frame into one exact-size buffer before any handler runs, so the frame goes back to its pool at once. A message read into an unpooled frame keeps that frame instead of copying it. SubZeroCopy() subscriptions alias the pooled frame and release it when the last handle is acked.

Session engine (internal/inflight). Owns packet identifiers, the send quota (the broker's Receive Maximum), outbound and inbound QoS 1/2 state machines, ordered acknowledgements, replay on resumption, and writes to the session.Store. Store writes run off the read goroutine and are drained on Disconnect.

Subscriptions. One broker subscription per exact filter, shared by the client's subscriptions with that filter; a topic trie maps topics to them, and Subscription Identifiers select the subscriptions a PUBLISH was sent for. SUBSCRIBE and UNSUBSCRIBE go through one ordered control queue that survives reconnects.

Code map

Path What
client.go Client, New, ServerInfo (what CONNACK granted), Stats counters
options.go Config, every With… option, backoff policies
lifecycle.go Connect/Disconnect, the supervisor, lifecycle callbacks and their queue, server redirects
conn.go, conn_unix.go, conn_other.go per-connection state: read loop and dispatch, writer goroutine, direct writes, keep-alive; the non-blocking socket write on unix
auth.go Authenticator interfaces, AUTH handling, Reauthenticate
publish.go Publish, broker-limit checks, publisher pool
subscribe.go, subscription.go subscribe API with its results and tokens, Message and its delivery, Queue[T]; the subscription registry and control queue
queue_publisher.go QueuePublisher, the PublisherQueue interface and its in-memory implementation
client_group.go ClientGroup
types.go public value types and errors
encoding.go Encoding and the typed calls: PublishValue, SubscribeAs and their variants, TypedMessage
doc.go package documentation
*_test.go a test file per source file, plus helpers_test.go, session_test.go (sessions and stores across connections), bench_test.go, example_test.go and readme_test.go
internal/inflight session engine: packet identifiers, QoS 1/2 state, quota, replay, store writes
internal/trie topic trie
internal/clock, internal/testbroker injectable clock; scripted and raw brokers for tests
internal/filedb bbolt wrapper with group commit, shared by store/file and queue/file
session, session/storetest Store interface, memory store, conformance tests for stores
queue/queuetest conformance tests for PublisherQueue implementations
wire MQTT v5 codec (unstable API)
transport, transport/ws TCP/TLS dialing; WebSocket
store/file, queue/file durable session store and publish queue
codec/json, codec/msgpack Encoding implementations
conformance tests against real brokers (-tags conformance)
benchmarks head-to-head benchmarks, raw results, benchtab
examples, examples/readme runnable examples, run against the conformance brokers by examples/smoke_test.go (-tags smoke); the README's snippets as compiled code

External dependencies

Module Depends on
github.com/ashtonian/mqttv5 (core) Go standard library only, tests included
store/file, queue/file go.etcd.io/bbolt through internal/filedb
transport/ws github.com/gobwas/ws
codec/json standard library
codec/msgpack github.com/vmihailenco/msgpack/v5
benchmarks eclipse/paho.golang, eclipse/paho.mqtt.golang, golang.org/x/perf (not a dependency of anything else)

At runtime the client needs only an MQTT v5 broker reachable over TCP, TLS or (with transport/ws) WebSocket.

Build and test

# Core — no broker required.
go test -race ./...

# Every module (each has its own go.mod).
for m in . codec/json codec/msgpack store/file queue/file internal/filedb transport/ws examples benchmarks; do
    (cd $m && go vet ./... && go test -race ./...)
done

# Conformance against mosquitto and EMQX (add `--profile hivemq` for
# HiveMQ CE; see conformance/README.md), then every example run against
# the same brokers.
docker compose -f conformance/docker-compose.yml up -d
go -C conformance test -tags conformance -race ./...
go -C examples test -tags smoke -v -run TestExamplesRun .

# Benchmarks: see benchmarks/README.md.
docker compose -f benchmarks/docker-compose.yml up -d
benchmarks/scripts/run.sh e2e . -tags e2e -run '^$' -bench '^BenchmarkE2E_' -benchmem -timeout 3h
go -C benchmarks run ./cmd/benchtab README.md ../README.md

Stability

mqttv5 is pre-1.0: any minor release may change the API.

  • The client API is the mqttv5 package: Client, Config, the options, Stats, and its value types (PublishOptions, TopicFilter, Message, TypedMessage, Properties, ConnackInfo, DisconnectInfo, WillOptions, ConnectOptions, DisconnectOptions, ServerRedirect, results and errors).
  • ReasonCode, PacketType and UserProperty are the protocol's own values, shared with the codec.
  • wire, the codec underneath, is exported for tools, test brokers and storage formats. The client API never requires it, and it may change in any release.
  • Submodules have their own go.mod and are tagged with the core at every release; use matching versions (Modules).

Independence

Independent, clean-room implementation written from the MQTT v5.0 OASIS specification. Not a fork of any existing Go MQTT client.

The benchmarks/ submodule imports eclipse/paho.golang for head-to-head comparison only — it is not redistributed.

License

Apache 2.0 — see LICENSE for the full text and NOTICE for the attribution notice. Per-file headers carry SPDX-License-Identifier: Apache-2.0.

Documentation

Overview

Package mqttv5 is an MQTT v5 client for Go that allocates little per message, with an ergonomic API. The supervisor (reconnect, replay-in-flight, auto-resubscribe) is baked into every Client; there is no separate "auto-reconnect" wrapper.

Why this package

One package, one Client. No paho / autopaho split — reconnect, replay, and resubscribe are always on. Channel-native subscribe: <-chan *Message (Client.Subscribe), Queue of *Message (Client.SubscribeQueue), or a sync callback (Client.SubscribeCallback). Backpressure is per subscription: a consumer that falls behind loses QoS 0 messages, acknowledged so the broker stops retransmitting, while the broker's flow control holds QoS 1 and 2 messages back unless the subscription opts out with SubAutoAck or a DropOldest queue (see Client.Subscribe).

Two multi-broker patterns are kept distinct:

Separately, WithPublisherPool adds publish-only connections to the client's own brokers, each with its own window of unacknowledged QoS 1/2 messages and its own read loop, while subscriptions and any persistent session stay on the main connection.

Typed publish / subscribe takes one Encoding per client (WithEncoding) and the payload type per call: Client.SubscribeAs delivers decoded values and Client.PublishValue encodes any value and declares the Encoding's Content Type. JSON and msgpack Encodings ship in separate submodules so the core stays stdlib-only. The durable outbound QueuePublisher lets you enqueue publishes while disconnected, keeps a window of them in flight once connected, and, with the queue/file and store/file submodules, continues the exchanges in flight after a process crash instead of publishing those messages again (a lost broker session aside).

With WithStore, every packet that depends on a session record is sent only once the record is written, and a failed write stops the client as a crash would: the next Client.Connect reloads the session from the store (see WithOnStoreFailure).

Operator surface:

The MQTT v5 client feature set: shared subscriptions, Subscription Identifiers, topic aliases (in and out), session expiry and resumption, retained messages, the Will and its properties, server redirects, enhanced authentication (CONNECT and mid-session, §4.12), and the broker's CONNACK limits enforced before a packet is sent. Every packet from the broker is validated before it is acted on.

Lifecycle callbacks

The connection callbacks (WithOnConnectionUp, WithOnConnectionDown, WithOnConnectError, WithOnReconnectAttempt, WithOnServerDisconnect, WithOnServerRedirect, WithOnResubscribeError, WithOnReauthenticated and WithOnStoreFailure) run one at a time, in the order of the events they report, on a goroutine of the client's own. They may call Client.Disconnect, Client.Connect and Client.SetBrokers. Before each reconnect attempt, and before Connect tries the next broker, the client waits for the callbacks posted so far, so a SetBrokers call from one applies to that attempt, and a slow callback delays it. Because they run apart from the calls that cause them, OnConnectionUp may run after Connect returns, and a callback for an event before Disconnect may run after Disconnect returns.

Message handlers are different: Client.SubscribeCallback handlers and SubOnDrop hooks run on the connection's read loop, which Disconnect waits for, so they must not call Disconnect themselves; start it on another goroutine. A teardown held up by one logs a warning after five seconds.

Quick start 

cli, err := mqttv5.New(
    mqttv5.WithBroker("mqtt://localhost:1883"),
    mqttv5.WithClientID("ingest-svc-1"),
)
if err != nil { panic(err) }

ctx := context.Background()
if err := cli.Connect(ctx); err != nil { panic(err) }
defer cli.Disconnect(ctx)

// Channel subscribe — manual ack, §4.6-ordered PUBACK flush.
msgs, _, err := cli.Subscribe(ctx,
    []mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}})
if err != nil { panic(err) }
go func() {
    for m := range msgs {
        fmt.Printf("%s: %s\n", m.Topic, m.Payload)
        _ = m.Ack()
    }
}()

// QoS 1 publish — supervisor replays with DUP=1 across drops.
_ = cli.Publish(ctx, PublishOptions{
    Topic:   "events/example",
    Payload: []byte("hello"),
    QoS:     1,
})

See the Example functions below for end-to-end snippets covering each subscribe shape, typed payloads, multi-broker fan-out, and the durable queue publisher.

Options naming

Option helpers split by scope: Option (passed to New) configures the Client and is named With…; SubscribeOption (passed inline to Subscribe / SubscribeQueue) configures one subscription and is named Sub…; ClientGroupOption (passed to NewClientGroup) configures the group and is named WithGroup….

Submodules

Each opt-in submodule has its own go.mod so importing it does not add a runtime dependency to the core:

  • codec/json — JSON [Codec] implementation, wired into [Typed].
  • codec/msgpack — MessagePack [Codec] via vmihailenco/msgpack/v5.
  • queue/file — Crash-safe outbound queue for QueuePublisher.
  • store/file — Crash-safe session store for in-flight QoS 1/2.
  • transport/ws — WebSocket transport via gobwas/ws; wire it in with WithDialFunc.

Architecture

One goroutine per connection drives the read path (read -> decode -> trie match -> handler) with handlers running synchronously on the reader. On a TCP or Unix connection, a publish that waits for its write writes on its own goroutine when the connection is idle; if its ctx ends mid-write, a deadline stops the write and the writer goroutine finishes the packet. When publishers overlap, and on TLS and WebSocket connections, a second goroutine drains a many-producer-single-consumer write channel and coalesces the queued packets, so they do not take turns on a write lock around net.Conn.Write. On unix systems the reader writes the acknowledgements it owes itself when the connection is idle, as far as the socket takes them without waiting, and leaves the rest to the writer goroutine. WithPublisherPool runs N such connections for publishing, each with its own writer and its own Receive Maximum window.

Packets and frame buffers up to 64 KiB come from sync.Pools. An inbound Message owns a copy of its topic, payload and properties, so a pooled frame goes back to its pool before any handler runs; a larger frame is never pooled and becomes the message's own. SubZeroCopy subscriptions alias the frame until every handle is acked.

A supervisor goroutine reconnects with configurable backoff, resumes the session, replays unacknowledged QoS 1/2 publishes with DUP=1 in their original order, and re-issues every tracked subscription.

Benchmarks

See benchmarks/ for end-to-end and codec benchmarks against eclipse/paho.golang and eclipse/paho.mqtt.golang.

Example

Example shows the minimal connect → subscribe → publish loop. The supervisor (reconnect, replay, resubscribe) is implicit — once Connect returns, every drop is recovered automatically until Disconnect. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, err := mqttv5.New(
		mqttv5.WithBroker("mqtt://localhost:1883"),
		mqttv5.WithClientID("example"),
	)
	if err != nil {
		panic(err)
	}

	ctx := context.Background()
	if err := cli.Connect(ctx); err != nil {
		panic(err)
	}
	defer cli.Disconnect(ctx)

	msgs, _, err := cli.Subscribe(ctx, []mqttv5.TopicFilter{{Topic: "demo/#", QoS: 1}})
	if err != nil {
		panic(err)
	}
	go func() {
		for m := range msgs {
			fmt.Printf("recv %s %s\n", m.Topic, m.Payload)
			_ = m.Ack()
		}
	}()

	_ = cli.Publish(ctx, mqttv5.PublishOptions{
		Topic:   "demo/hello",
		Payload: []byte("world"),
		QoS:     1,
	})
}

Index

Examples

Constants

View Source 
const (
	// DefaultKeepAlive is the keepalive interval used when
	// WithKeepAlive is not called. MQTT v5 spec range is 1..65535
	// seconds.
	DefaultKeepAlive uint16 = 30

	// DefaultConnectTimeout bounds the CONNECT handshake (dial +
	// CONNECT/CONNACK) when WithConnectTimeout is not called.
	DefaultConnectTimeout = 10 * time.Second

	// DefaultPingTimeout caps how long the client waits for any packet
	// after a PINGREQ before declaring the connection dead. The default
	// is the smaller of this and half the keep-alive, so a dead
	// connection is always detected before the next PINGREQ would be
	// due. With DefaultKeepAlive (30 s) the dead-detect time is 40 s,
	// inside a broker's 1.5×KeepAlive (45 s) disconnect window.
	DefaultPingTimeout = 10 * time.Second

	// DefaultSessionExpiry is the Session Expiry Interval (§3.1.2.11.2)
	// advertised on every CONNECT. 5 minutes is long enough to ride out
	// the typical reconnect blip / container redeploy window so QoS 1/2
	// resume actually works out of the box; short enough that the
	// broker isn't asked to hold idle sessions indefinitely.
	DefaultSessionExpiry uint32 = 300

	// DefaultWriteQueueSize sizes the internal MPSC write channel.
	DefaultWriteQueueSize = 256

	// DefaultDisconnectFlushTimeout caps how long writeDisconnectAndClose
	// will wait for the DISCONNECT packet to flush before the
	// connection is forcibly torn down.
	DefaultDisconnectFlushTimeout = 500 * time.Millisecond

	// DefaultReadBufferSize is the per-connection read window used when
	// [WithReadBufferSize] is not called. 16 KiB cuts read syscalls for
	// streams of small messages fourfold against 4 KiB; larger windows
	// gain little (BenchmarkReceiveWindow).
	DefaultReadBufferSize = 16 << 10

	// DefaultMaxSubscribeQueueSize caps each [Client.SubscribeQueue]
	// when no cap is configured. At roughly 200 B of overhead plus the
	// payload per queued message, a full default queue of small
	// messages holds about 13 MiB plus payloads.
	DefaultMaxSubscribeQueueSize = 65536

	// DefaultReceiveMaximum is the Receive Maximum the client advertises
	// when [WithReceiveMaximum] is not called.
	DefaultReceiveMaximum = 256

	// UnboundedQueue, passed to [WithMaxSubscribeQueueSize] or
	// [SubMaxQueueSize], removes the queue cap. A consumer that falls
	// behind then grows memory without limit.
	UnboundedQueue = -1
)

Default values used by Config.defaults. Exported so callers can reference them when overriding selectively.

View Source 
const (
	ReasonSuccess                           = wire.ReasonSuccess
	ReasonNormalDisconnection               = wire.ReasonNormalDisconnection
	ReasonGrantedQoS0                       = wire.ReasonGrantedQoS0
	ReasonGrantedQoS1                       = wire.ReasonGrantedQoS1
	ReasonGrantedQoS2                       = wire.ReasonGrantedQoS2
	ReasonDisconnectWithWill                = wire.ReasonDisconnectWithWill
	ReasonNoMatchingSubscribers             = wire.ReasonNoMatchingSubscribers
	ReasonNoSubscriptionExisted             = wire.ReasonNoSubscriptionExisted
	ReasonContinueAuthentication            = wire.ReasonContinueAuthentication
	ReasonReAuthenticate                    = wire.ReasonReAuthenticate
	ReasonUnspecifiedError                  = wire.ReasonUnspecifiedError
	ReasonMalformedPacket                   = wire.ReasonMalformedPacket
	ReasonProtocolError                     = wire.ReasonProtocolError
	ReasonImplementationSpecificError       = wire.ReasonImplementationSpecificError
	ReasonUnsupportedProtocolVersion        = wire.ReasonUnsupportedProtocolVersion
	ReasonClientIdentifierNotValid          = wire.ReasonClientIdentifierNotValid
	ReasonBadUsernameOrPassword             = wire.ReasonBadUsernameOrPassword
	ReasonNotAuthorized                     = wire.ReasonNotAuthorized
	ReasonServerUnavailable                 = wire.ReasonServerUnavailable
	ReasonServerBusy                        = wire.ReasonServerBusy
	ReasonBanned                            = wire.ReasonBanned
	ReasonServerShuttingDown                = wire.ReasonServerShuttingDown
	ReasonBadAuthenticationMethod           = wire.ReasonBadAuthenticationMethod
	ReasonKeepAliveTimeout                  = wire.ReasonKeepAliveTimeout
	ReasonSessionTakenOver                  = wire.ReasonSessionTakenOver
	ReasonTopicFilterInvalid                = wire.ReasonTopicFilterInvalid
	ReasonTopicNameInvalid                  = wire.ReasonTopicNameInvalid
	ReasonPacketIdentifierInUse             = wire.ReasonPacketIdentifierInUse
	ReasonPacketIdentifierNotFound          = wire.ReasonPacketIdentifierNotFound
	ReasonReceiveMaximumExceeded            = wire.ReasonReceiveMaximumExceeded
	ReasonTopicAliasInvalid                 = wire.ReasonTopicAliasInvalid
	ReasonPacketTooLarge                    = wire.ReasonPacketTooLarge
	ReasonMessageRateTooHigh                = wire.ReasonMessageRateTooHigh
	ReasonQuotaExceeded                     = wire.ReasonQuotaExceeded
	ReasonAdministrativeAction              = wire.ReasonAdministrativeAction
	ReasonPayloadFormatInvalid              = wire.ReasonPayloadFormatInvalid
	ReasonRetainNotSupported                = wire.ReasonRetainNotSupported
	ReasonQoSNotSupported                   = wire.ReasonQoSNotSupported
	ReasonUseAnotherServer                  = wire.ReasonUseAnotherServer
	ReasonServerMoved                       = wire.ReasonServerMoved
	ReasonSharedSubscriptionsNotSupported   = wire.ReasonSharedSubscriptionsNotSupported
	ReasonConnectionRateExceeded            = wire.ReasonConnectionRateExceeded
	ReasonMaximumConnectTime                = wire.ReasonMaximumConnectTime
	ReasonSubscriptionIDsNotSupported       = wire.ReasonSubscriptionIDsNotSupported
	ReasonWildcardSubscriptionsNotSupported = wire.ReasonWildcardSubscriptionsNotSupported
)

Reason codes, as named by MQTT v5 §2.4. Several share a value; the packet they arrive in decides the meaning.

View Source 
const (
	CONNECT     = wire.CONNECT
	CONNACK     = wire.CONNACK
	PUBLISH     = wire.PUBLISH
	PUBACK      = wire.PUBACK
	PUBREC      = wire.PUBREC
	PUBREL      = wire.PUBREL
	PUBCOMP     = wire.PUBCOMP
	SUBSCRIBE   = wire.SUBSCRIBE
	SUBACK      = wire.SUBACK
	UNSUBSCRIBE = wire.UNSUBSCRIBE
	UNSUBACK    = wire.UNSUBACK
	PINGREQ     = wire.PINGREQ
	PINGRESP    = wire.PINGRESP
	DISCONNECT  = wire.DISCONNECT
	AUTH        = wire.AUTH
)

Control packet types.

View Source 
const DefaultPublisherPoolClientIDFormat = "%s-pub-%d"

DefaultPublisherPoolClientIDFormat is the format string used by the default pool-member ClientID function: parent ClientID, then "-pub-", then the 1-based member index.

View Source 
const (
	// DefaultQueueWindow is how many queued messages a QueuePublisher
	// publishes at once when [WithQueueWindow] is not set.
	DefaultQueueWindow = 32
)
View Source 
const DefaultSubscribeBuffer = 64

DefaultSubscribeBuffer is the channel buffer size used by Subscribe when SubBuffer is not called: how many QoS 0 messages the channel holds before it drops one.

View Source 
const QueueIDProperty = "mqttv5-msg-id"

QueueIDProperty is the user property WithQueueIdempotencyKey adds to each queued message, holding its QueueEntry.ID.

Variables

View Source 
var (
	// ErrQueueClosed is returned by a [PublisherQueue] or
	// [QueuePublisher] after Close.
	ErrQueueClosed = errors.New("mqttv5: queue is closed")

	// ErrQueueFull is returned when a bounded queue has no room: under
	// [DropNewest] when it holds its maximum, under [DropOldest] when
	// every entry it holds is already being published.
	ErrQueueFull = errors.New("mqttv5: queue is full")
)
View Source 
var (
	ErrNotConnected     = errors.New("mqttv5: client not connected")
	ErrAlreadyConnected = errors.New("mqttv5: client already connected")
	ErrClosed           = errors.New("mqttv5: client closed")
	ErrConnectRefused   = errors.New("mqttv5: broker refused CONNECT")
	ErrUnexpectedPacket = errors.New("mqttv5: unexpected packet from broker")

	// Enhanced-authentication errors (§4.12), returned by Reauthenticate.
	// ErrReauthRejected is wrapped with the broker's DISCONNECT reason
	// code (mirroring ErrConnectRefused) — match with errors.Is.
	ErrNoAuthenticator  = errors.New("mqttv5: no Authenticator configured")
	ErrReauthInProgress = errors.New("mqttv5: re-authentication already in progress")
	ErrReauthRejected   = errors.New("mqttv5: broker rejected re-authentication")

	// ErrWriteQueueFull is returned by QoS 0 Publish when the client
	// is configured with WriteDropNewest and the writer queue has no
	// room. The publish never reaches the wire.
	ErrWriteQueueFull = errors.New("mqttv5: write queue full")

	// Subscribe-time errors raised when the broker advertised the
	// feature as unsupported in CONNACK (MQTT v5 §3.2.2.3.{11,13}).
	// Returned before any SUBSCRIBE traffic goes on the wire.
	ErrSharedSubsUnsupported   = errors.New("mqttv5: broker does not support shared subscriptions")
	ErrWildcardSubsUnsupported = errors.New("mqttv5: broker does not support wildcard subscriptions")

	// ErrSessionLost completes a QoS 1/2 Publish whose message was
	// discarded because a connection started without the old session
	// and [WithSessionLossPolicy] is [SessionLossFail].
	ErrSessionLost = inflight.ErrSessionLost

	// ErrMessageExpired completes a QoS 1/2 Publish whose Message Expiry
	// Interval ran out before it could be sent again after a session
	// loss.
	ErrMessageExpired = inflight.ErrMessageExpired

	// ErrPacketIDsExhausted is returned when all 65,535 packet
	// identifiers are in use and the caller's context ends first.
	ErrPacketIDsExhausted = inflight.ErrIDsExhausted

	// ErrStoreFailed is matched by every [*StoreError]: a write to the
	// [WithStore] store failed.
	ErrStoreFailed = inflight.ErrStoreFailed

	// ErrQoSNotSupported is returned by Publish for a QoS above the
	// broker's Maximum QoS (see [WithQoSDowngrade]), and completes a
	// stored message that a reconnect to such a broker can no longer
	// send.
	ErrQoSNotSupported = inflight.ErrQoSNotSupported

	// ErrInvalidSessionExpiry is returned by [Client.DisconnectWith]
	// for a non-zero Session Expiry Interval when the CONNECT carried 0
	// [MQTT-3.14.2-2]. Nothing is sent.
	ErrInvalidSessionExpiry = errors.New("mqttv5: DISCONNECT cannot set a Session Expiry Interval when CONNECT sent 0")

	// ErrTopicAliasInvalid is returned by Publish for a Topic Alias the
	// broker's Topic Alias Maximum does not allow, or for a QoS 1/2
	// message that has an alias but no topic name.
	ErrTopicAliasInvalid = errors.New("mqttv5: invalid topic alias")

	// ErrInvalidTopic is returned for a topic name or filter MQTT v5
	// forbids (§4.7): wildcards in a topic name, misplaced wildcards in
	// a filter, a malformed shared subscription, NUL or invalid UTF-8.
	ErrInvalidTopic = wire.ErrInvalidTopic

	// ErrInvalidField is returned for any other option value MQTT v5
	// forbids, such as No Local on a shared subscription or DUP on a
	// QoS 0 publish. Nothing is sent.
	ErrInvalidField = wire.ErrInvalidField

	// ErrFieldTooLong is returned for a string or binary field over the
	// 65,535 bytes the protocol can encode.
	ErrFieldTooLong = wire.ErrStringTooLong

	// ErrRetainNotSupported is returned by Publish for a retained
	// message when the broker's Retain Available is 0.
	ErrRetainNotSupported = inflight.ErrRetainNotSupported

	// ErrPacketTooLarge is returned for a packet larger than the
	// broker's Maximum Packet Size, and matches a broker refusal with
	// reason code 0x95.
	ErrPacketTooLarge = inflight.ErrPacketTooLarge

	// Sentinels matched by [ReasonCodeError] for the broker refusals
	// callers most often branch on.
	ErrNotAuthorized        = errors.New("mqttv5: not authorized")
	ErrQuotaExceeded        = errors.New("mqttv5: quota exceeded")
	ErrTopicNameInvalid     = errors.New("mqttv5: topic name invalid")
	ErrPayloadFormatInvalid = errors.New("mqttv5: payload format invalid")
)

Errors returned by the client.

View Source 
var DefaultQueueRetryBackoff = ExponentialBackoff(time.Second, time.Minute, 500*time.Millisecond)

DefaultQueueRetryBackoff spaces the attempts of a message whose publish failed for a reason that may pass (a refusal such as 0x97 Quota exceeded, a store error).

View Source 
var DefaultReconnectBackoff = ExponentialBackoff(
	time.Second, 30*time.Second, 200*time.Millisecond,
)

DefaultReconnectBackoff is used when WithReconnectBackoff is not called: 1s -> 2s -> 4s -> ... up to 30s with 200ms jitter.

View Source 
var ErrChanDropOldestUnsupported = errors.New(
	"mqttv5: Subscribe (chan) does not support DropOldest; use SubscribeQueue",
)

ErrChanDropOldestUnsupported is returned by Client.Subscribe when called with SubDropPolicy(DropOldest); use Client.SubscribeQueue for DropOldest semantics.

View Source 
var ErrInvalidBrokerURL = errors.New("mqttv5: invalid broker URL")

ErrInvalidBrokerURL is returned when a broker URL fails to parse or has an unsupported scheme. WithDialFunc relaxes scheme validation.

View Source 
var ErrMissingBroker = errors.New("mqttv5: missing broker URL (use WithBroker)")

ErrMissingBroker is returned by New when no broker URL was provided.

View Source 
var ErrNilHandler = errors.New("mqttv5: subscribe handler must not be nil")

ErrNilHandler is returned by Client.SubscribeCallback when the handler argument is nil.

View Source 
var ErrNoEncoding = errors.New("mqttv5: no Encoding: set WithEncoding, or SubEncoding for one subscription")

ErrNoEncoding is returned by a typed call on a client without an Encoding: set one with WithEncoding, or with SubEncoding for one subscription.

View Source 
var ErrQoS0NotQueueable = errors.New("mqttv5: QueuePublisher requires QoS >= 1")

ErrQoS0NotQueueable is returned by QueuePublisher.Publish for a QoS 0 message: with no acknowledgement there is nothing to keep it queued for.

Functions

func EncodePublish added in v0.11.0 

func EncodePublish(o PublishOptions, id uint16) ([]byte, error)

EncodePublish encodes o as a PUBLISH packet with packet identifier id (ignored for QoS 0). For tools and storage formats that need the message's exact MQTT encoding, such as a PublisherQueue.

func PermanentPublishError added in v0.11.0 

func PermanentPublishError(err error) bool

PermanentPublishError reports whether err means the broker will never accept the message as it is: PUBACK/PUBREC 0x87 Not authorized, 0x90 Topic Name invalid, 0x95 Packet too large, 0x99 Payload format invalid, 0x9A Retain not supported, 0x9B QoS not supported, or the equivalent checks against the broker's CONNACK limits and MQTT's rules before sending. 0x97 Quota exceeded and other failures may pass.

Types

type Authenticator 

type Authenticator interface {
	Method() string
	Begin(ctx context.Context) (data []byte, err error)
	Continue(brokerData []byte) (response []byte, done bool, err error)
}

Authenticator drives the client side of MQTT v5 enhanced authentication — the challenge-response handshake that runs during CONNECT and during re-authentication (SCRAM, Kerberos, OAuth/JWT, etc.).

Method() supplies the AuthenticationMethod property. Begin(ctx) resolves the initial AuthenticationData — for a bearer method this is where a fresh token is fetched, so ctx carries the deadline of the triggering operation (the connect timeout, or the ctx passed to Reauthenticate); honor it for any I/O. For each broker AUTH challenge — during CONNECT (before CONNACK) and during mid-session re-authentication (§4.12) — Continue(brokerData) returns the next response, which the client always sends as AUTH 0x18 Continue.

done is informational only. The client never concludes the exchange: per §3.15.2.1 only the server sends AUTH 0x00 Success (or CONNACK for the initial handshake). Returning done=true does NOT make the client emit 0x00 — treat it as a hint that the Authenticator has no further data to send. A non-nil error from Continue aborts the connection.

Implementations need only be safe for concurrent use when shared across multiple Clients (e.g. with WithPublisherPool).

type Backoff 

type Backoff func(attempt int) time.Duration

Backoff returns the delay before retry attempt N. The first retry after a failed connect is attempt 0 (the initial connect itself runs immediately, no backoff). Implementations must be safe for concurrent use.

func ConstantBackoff 

func ConstantBackoff(d time.Duration) Backoff

ConstantBackoff returns the same delay for every attempt. Predictable but vulnerable to thundering-herd reconnect storms; ExponentialBackoff is usually preferable.

func ExponentialBackoff 

func ExponentialBackoff(min, max, jitter time.Duration) Backoff

ExponentialBackoff doubles min each attempt up to max with ±jitter added (range [computed-jitter, computed+jitter]).

type Client 

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

Client is an MQTT v5 client. After New(opts...), call Connect to establish the first connection and start the supervisor that reconnects on drop.

func New 

func New(opts ...Option) (*Client, error)

New constructs a Client. Apply WithBroker (required) plus any other options. The Client is not connected — call Connect.

func (*Client) AwaitConnection added in v0.11.0 

func (c *Client) AwaitConnection(ctx context.Context) error

AwaitConnection waits until the client is connected and returns nil, at once if it already is. It returns ErrClosed when the client is disconnected, or ctx's error. Pair with WithRetryInitialConnect to start without the broker and wait for it.

func (*Client) ClientID 

func (c *Client) ClientID() string

ClientID returns the ClientID currently in effect: the one configured with WithClientID, or, when that was empty, the identifier the broker assigned (AssignedClientIdentifier, §3.2.2.3.7). The assigned identifier is reused on every later CONNECT so the session can resume, and is kept in the WithStore store across restarts.

func (*Client) Connect 

func (c *Client) Connect(ctx context.Context) error

Connect performs the initial CONNECT/CONNACK handshake (bounded by ctx) and starts the supervisor. With WithBrokers it tries each broker once, from a random one on, and fails only when none accepts, with every broker's error. Reconnect attempts run on internal background contexts until Client.Disconnect. With WithRetryInitialConnect a failed handshake is retried the same way and Connect returns nil; see Client.AwaitConnection. Returns ErrAlreadyConnected when called twice without an intervening Disconnect; while a Disconnect, or a supervisor told to stop reconnecting, is still tearing down, Connect waits for it. A Disconnect during Connect cancels it, and Connect returns ErrClosed.

func (*Client) Connected 

func (c *Client) Connected() bool

Connected reports whether the client currently has a live connection.

func (*Client) Disconnect 

func (c *Client) Disconnect(ctx context.Context) error

Disconnect sends a graceful DISCONNECT (reason Normal Disconnection), stops the supervisor, and tears down the current connection. For a custom reason / properties use Client.DisconnectWith. Does not fire OnConnectionDown. Idempotent.

Lifecycle callbacks may call Disconnect. Client.SubscribeCallback handlers and SubOnDrop hooks must not: Disconnect waits for the read loop that runs them, so it would wait for itself. Start it on another goroutine.

func (*Client) DisconnectWith 

func (c *Client) DisconnectWith(ctx context.Context, opts DisconnectOptions) error

DisconnectWith sends a DISCONNECT with opts, stops the supervisor, and tears down the current connection. With WithPublisherPool it then refuses new publishes with ErrClosed and waits, bounded by ctx, for those in flight on pool members, whose sessions end with their connections, before disconnecting them. It returns ctx's error if ctx ended before the session store writes or the pool's publishes were done, joined with the pool members' disconnect errors. Idempotent. Like Client.Disconnect, it must not be called from a SubscribeCallback handler or SubOnDrop hook.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, po PublishOptions) error

Publish sends a PUBLISH packet.

QoS 1/2 messages belong to the session once registered: a dropped connection resends them (DUP=1) when the session resumes, and callers stay blocked on the acknowledgement across the drop. A refusal (reason code 0x80 or above) is returned as a *ReasonCodeError.

A call that waits for its packet to be written — QoS 1/2, or QoS 0 with PublishWaitForFlush — writes it on the calling goroutine when the connection is a TCP or Unix socket and nothing is queued for its writer goroutine, saving the hand-off; otherwise the writer sends it behind the queued packets. ctx bounds the call either way — a write still blocked when ctx is cancelled stops within 20 ms, one at ctx's deadline at once — and a write it interrupts is finished by the writer goroutine, so the packet may still reach the broker after Publish returned ctx's error.

With WithPublisherPool the call first tries the pool per PoolRoutingPolicy and falls back to the main connection if every pool member is unhealthy.

Example

ExampleClient_Publish sends a QoS 1 publish. Publish returns after PUBACK arrives. If the connection drops mid-flight the session entry persists and the supervisor replays the packet with DUP=1 on reconnect; the caller stays blocked across the drop. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, _ := mqttv5.New(mqttv5.WithBroker("mqtt://localhost:1883"))
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	err := cli.Publish(ctx, mqttv5.PublishOptions{
		Topic:   "events/example",
		Payload: []byte(`{"value":42}`),
		QoS:     1,
	})
	if err != nil {
		fmt.Println("publish:", err)
	}
}

func (*Client) PublishValue added in v0.12.0 

func (c *Client) PublishValue(ctx context.Context, opts PublishOptions, v any) error

PublishValue encodes v with the client's Encoding and publishes it as Client.Publish does, declaring the Encoding's Content Type unless opts sets one. opts.Payload must be nil: v is the payload.

func (*Client) Reauthenticate added in v0.10.0 

func (c *Client) Reauthenticate(ctx context.Context) error

Reauthenticate initiates MQTT v5 re-authentication (§4.12) on each live connection the client holds: the main one and, with WithPublisherPool, every connected pool member, all at once. On each connection it sends an AUTH 0x19 carrying the configured Authenticator's Method() and a fresh Begin(ctx) payload, services any broker 0x18 Continue challenges via Authenticator.Continue, and returns when the broker concludes the exchange — without dropping the connection or disturbing in-flight QoS state.

It returns nil when the broker accepts re-authentication (AUTH 0x00 Success), or:

  • ErrNoAuthenticator if the client was not built WithAuthenticator;
  • ErrNotConnected if there is no live connection (before the first CONNACK, or after a drop);
  • ErrReauthInProgress if another re-auth is already in flight on this connection (calls are single-flighted per connection);
  • ErrReauthRejected (wrapped with the DISCONNECT reason code) if the broker rejects re-auth and tears the connection down;
  • ctx.Err() if ctx is cancelled or its deadline elapses first.

The AuthenticationMethod established at CONNECT is reused; Method() MUST be stable for the client's lifetime (§4.12 requires it to match across the exchange). Reauthenticate is safe for concurrent use.

ctx bounds the whole operation including credential resolution via Begin(ctx). On ctx cancellation after the AUTH 0x19 has been sent the connection is left intact and the exchange resolves on the wire — re-auth stays single-flighted until the broker concludes, so a subsequent call may observe ErrReauthInProgress until then. On a broker rejection the connection is torn down and the supervisor reconnects through the normal CONNECT path (which re-presents credentials).

With a publisher pool the outcomes above describe the main connection, joined with any pool member's failure (prefixed with the member), so errors.Is sees every one: a rejected member reconnects as the main connection would, and one lost during its exchange reports ErrNotConnected. A member not connected when the call starts is skipped; its next CONNECT presents a fresh credential.

func (*Client) ServerInfo added in v0.11.0 

func (c *Client) ServerInfo() (ConnackInfo, bool)

ServerInfo returns what the broker granted on the current connection, and false when the client is not connected.

func (*Client) SetBrokers 

func (c *Client) SetBrokers(urls ...string) error

SetBrokers replaces the broker URL list at runtime, for the main connection and every publisher pool member; each one's next reconnect uses it. Typical caller: WithOnServerDisconnect on a ServerMoved / UseAnotherServer redirect. URLs are validated as in New.

Example

ExampleClient_SetBrokers shows multi-broker failover. The supervisor rotates through mqttv5.Config.BrokerURLs on every reconnect attempt; mqttv5.Client.SetBrokers replaces the list at runtime — the canonical hook is a ServerMoved DISCONNECT. 

package main

import (
	"crypto/tls"

	"github.com/ashtonian/mqttv5"
)

func main() {
	var cli *mqttv5.Client
	cli, _ = mqttv5.New(
		mqttv5.WithBrokers(
			"mqtts://broker-a.example.com:8883",
			"mqtts://broker-b.example.com:8883",
		),
		mqttv5.WithTLSConfig(&tls.Config{MinVersion: tls.VersionTLS13}),
		mqttv5.WithOnServerDisconnect(func(d mqttv5.DisconnectInfo) {
			if d.ServerReference != "" {
				_ = cli.SetBrokers(d.ServerReference)
			}
		}),
	)
	_ = cli
}

func (*Client) Stats 

func (c *Client) Stats() Stats

Stats returns a snapshot of the Client's counters, its publisher pool's included. Cheap to call — the returned value is detached. Returns the zero value when stats are disabled (WithStats not set on the Config).

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (<-chan *Message, SubscriptionToken, error)

Subscribe sends one SUBSCRIBE for filters and returns a buffered channel of matching inbound messages. Callers MUST call Message.Ack on each. Returns once the SUBACK arrives — if the client is not connected, or the connection drops first, the SUBSCRIBE goes out on the next connection — or when ctx ends. The channel closes on Client.Unsubscribe or Client.Disconnect, or when the broker refuses every filter.

Called before Client.Connect, or between Disconnect and the next Connect, Subscribe returns at once and the SUBSCRIBE goes out first on the connection Connect makes: messages a resumed session holds, which the broker sends right after its CONNACK, reach the subscription. token.Results() and token.Err() report the SUBACK once it arrives; if that Connect fails, the subscription ends with it.

When the broker refuses filters Subscribe returns a *SubscribeError together with the channel and token: the channel carries the granted filters, or is already closed if none were. token.Results() holds each filter's outcome.

The broker keeps one subscription per filter: subscriptions of this client with the same filter share it, each receives every message for it, and the options of the most recent Subscribe apply to all of them.

A consumer that falls behind loses QoS 0 messages, never QoS 1 or 2: once the channel holds SubBuffer messages, a QoS 0 message is dropped and acked (observe drops via SubOnDrop), while QoS 1 and 2 messages have room for WithReceiveMaximum more. That is as many as the broker sends before the consumer acks one, so the broker's flow control holds the rest back. The exceptions: with SubAutoAck, QoS 1 and 2 messages are dropped as QoS 0 ones are, and messages left unread from a session the broker discarded take room of their own. Returns ErrChanDropOldestUnsupported when called with explicit SubDropPolicy of DropOldest.

Example

ExampleClient_Subscribe receives matched messages on a buffered channel. The caller MUST call mqttv5.Message.Ack on each one — for QoS 1/2 the broker-side ack is held until Ack is called, in §4.6 arrival order across overlapping filters. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, _ := mqttv5.New(mqttv5.WithBroker("mqtt://localhost:1883"))
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	msgs, token, err := cli.Subscribe(ctx,
		[]mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}},
		mqttv5.SubBuffer(256),
		mqttv5.SubOnDrop(func(m *mqttv5.Message) {
			// Dropped messages have already been ack'd; this is for
			// observability only. Must not call Ack and must not retain
			// Topic/Payload past return.
			fmt.Println("dropped", m.Topic)
		}),
	)
	if err != nil {
		return
	}
	for m := range msgs {
		fmt.Println(m.Topic, string(m.Payload))
		_ = m.Ack()
	}
	_ = cli.Unsubscribe(ctx, token)
}

func (*Client) SubscribeAs added in v0.12.0 

func (c *Client) SubscribeAs[T any](ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (<-chan *TypedMessage[T], SubscriptionToken, error)

SubscribeAs is Client.Subscribe with each payload decoded into a T by the subscription's Encoding: SubEncoding, else the client's. Decoding runs on the read goroutine as messages arrive; a payload the Encoding rejects is not delivered (see SubOnDecodeError). Buffering, drops, refusals and closing are those of Client.Subscribe, and each message must be acked.

Example

ExampleClient_SubscribeAs sets one mqttv5.Encoding on the client and names the payload type per subscription. PublishValue encodes any value and declares the Encoding's Content Type. 

package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

// jsonEncoding is a tiny inline [mqttv5.Encoding] used by
// [ExampleClient_SubscribeAs]. In real code use the codec/json
// submodule's Encoding.
type jsonEncoding struct{}

func (jsonEncoding) ContentType() string             { return "application/json" }
func (jsonEncoding) Marshal(v any) ([]byte, error)   { return json.Marshal(v) }
func (jsonEncoding) Unmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }

type reading struct {
	Device string  `json:"device"`
	Temp   float64 `json:"temp"`
}

type command struct {
	Op string `json:"op"`
}

func main() {
	cli, _ := mqttv5.New(
		mqttv5.WithBroker("mqtt://localhost:1883"),
		mqttv5.WithEncoding(jsonEncoding{}),
	)
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	readings, _, err := cli.SubscribeAs[reading](ctx,
		[]mqttv5.TopicFilter{{Topic: "sensors/+", QoS: 1}})
	if err != nil {
		return
	}
	commands, _, err := cli.SubscribeAs[command](ctx,
		[]mqttv5.TopicFilter{{Topic: "devices/+/cmd", QoS: 1}})
	if err != nil {
		return
	}

	_ = cli.PublishValue(ctx,
		mqttv5.PublishOptions{Topic: "sensors/a1", QoS: 1},
		reading{Device: "a1", Temp: 22.5},
	)

	for {
		select {
		case m := <-readings:
			fmt.Println(m.Topic, m.Value.Temp)
			_ = m.Ack()
		case m := <-commands:
			fmt.Println(m.Topic, m.Value.Op)
			_ = m.Ack()
		}
	}
}

func (*Client) SubscribeCallback 

func (c *Client) SubscribeCallback(ctx context.Context, filters []TopicFilter, h HandlerFunc, opts ...SubscribeOption) (SubscriptionToken, error)

SubscribeCallback registers a subscription that invokes h for every matching inbound PUBLISH.

h runs synchronously on the read goroutine, so keep it short: while it runs the connection reads nothing, and other subscriptions' messages and the acknowledgements of the client's own publishes wait behind it (the keep-alive does not take that wait for a dead broker). It must not call Client.Disconnect, which waits for the read goroutine: start it on another goroutine. The runtime auto-acks via Message.Ack after h returns; use Client.Subscribe or Client.SubscribeQueue when ack must be deferred past h.

Of the [SubscribeOption]s only SubZeroCopy applies; with it, the message's fields are valid only until h returns. Refusals, reconnects, and a call before Connect behave as for Client.Subscribe. Returns ErrNilHandler when h is nil.

Example

ExampleClient_SubscribeCallback registers a synchronous handler invoked on the read goroutine. The handler must not block — a slow callback stalls PINGRESP processing and drops the connection. Ack fires automatically after the callback returns. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, _ := mqttv5.New(mqttv5.WithBroker("mqtt://localhost:1883"))
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	_, err := cli.SubscribeCallback(ctx,
		[]mqttv5.TopicFilter{{Topic: "ctrl/+", QoS: 0}},
		func(m *mqttv5.Message) {
			fmt.Println(m.Topic, string(m.Payload))
		},
	)
	if err != nil {
		return
	}
}

func (*Client) SubscribeCallbackAs added in v0.12.0 

func (c *Client) SubscribeCallbackAs[T any](ctx context.Context, filters []TopicFilter, h func(*TypedMessage[T]), opts ...SubscribeOption) (SubscriptionToken, error)

SubscribeCallbackAs is Client.SubscribeCallback with each payload decoded into a T, as Client.SubscribeAs decodes it. h runs on the read goroutine and the message is acked when it returns.

func (*Client) SubscribeQueue 

func (c *Client) SubscribeQueue(ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (*Queue[*Message], SubscriptionToken, error)

SubscribeQueue sends one SUBSCRIBE and returns a Queue of matching messages, capped at DefaultMaxSubscribeQueueSize unless WithMaxSubscribeQueueSize or SubMaxQueueSize say otherwise. On overflow, SubDropPolicy decides: DropNewest acks and drops the inbound message, except that QoS 1 and 2 messages have room for WithReceiveMaximum more beyond the cap, as on Client.Subscribe; DropOldest dequeues and acks the head, whatever its QoS, and admits the new one. Closes on Client.Unsubscribe or Client.Disconnect. Refusals, reconnects, and a call before Connect behave as for Client.Subscribe.

Example

ExampleClient_SubscribeQueue stores matched messages in an unbounded queue. mqttv5.DropOldest evicts the queue head when SubMaxQueueSize is reached so the freshest N messages survive. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, _ := mqttv5.New(mqttv5.WithBroker("mqtt://localhost:1883"))
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	q, _, err := cli.SubscribeQueue(ctx,
		[]mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}},
		mqttv5.SubMaxQueueSize(10_000),
		mqttv5.SubDropPolicy(mqttv5.DropOldest),
	)
	if err != nil {
		return
	}
	for {
		m, ok := q.Dequeue(ctx)
		if !ok {
			break
		}
		fmt.Println(m.Topic, string(m.Payload))
		_ = m.Ack()
	}
}

func (*Client) SubscribeQueueAs added in v0.12.0 

func (c *Client) SubscribeQueueAs[T any](ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (*Queue[*TypedMessage[T]], SubscriptionToken, error)

SubscribeQueueAs is Client.SubscribeQueue with each payload decoded into a T, as Client.SubscribeAs decodes it.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ctx context.Context, token SubscriptionToken) error

Unsubscribe ends the subscription: its channel or queue is closed before Unsubscribe returns, and an UNSUBSCRIBE is sent for every filter no other subscription of this client uses — the broker holds one subscription per filter, shared by all of them. Unsubscribe then waits for the broker's answer, across reconnects, until ctx ends; the UNSUBSCRIBE is still sent after that. Refusals are returned as an *UnsubscribeError.

type ClientGroup 

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

ClientGroup manages N parallel sessions to N independent brokers. Each member is a full Client with its own session and supervisor; the group adds dispatch policy on top. Use for bridges across independent brokers, multi-tenant SaaS with per-broker credentials, or devices split across independent clusters. For HA failover use WithBrokers on a single Client instead — ClientGroup does not fail over between members, though a member with several GroupMember.Brokers fails over among them. For publish throughput into one broker or cluster use WithPublisherPool: members of a group over one cluster would each receive every message a merged subscription to an unshared filter matches. A group is used once: after Disconnect its methods return ErrClosed.

func NewClientGroup 

func NewClientGroup(members []GroupMember, opts ...ClientGroupOption) (*ClientGroup, error)

NewClientGroup constructs a ClientGroup over the given members. Each GroupMember becomes a full *Client; the group applies the configured publish policy on top.

Member ClientIDs are kept unique: a member whose ClientID another already uses gets the first free "-N" suffix (N from 2); members without one get one each from their broker. Set GroupMember.Opts with WithClientID for explicit per-member IDs.

Defaults: GroupPublishBroadcast, parallel Connect/Disconnect.

Example

ExampleNewClientGroup creates one mqttv5.Client per broker and dispatches via the configured GroupPublishPolicy. Subscribe merges inbound messages from every member into one channel.

For failover within one logical client (interchangeable brokers, same data), use mqttv5.WithBrokers on a single Client instead — see [Example_ClientSetBrokers]. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	group, err := mqttv5.NewClientGroup(
		[]mqttv5.GroupMember{
			{
				Brokers: []string{"mqtts://emea.example.com:8883"},
				Name:    "emea",
				Opts:    []mqttv5.Option{mqttv5.WithCredentials("emea-svc", []byte("token-1"))},
			},
			{
				Brokers: []string{"mqtts://apac.example.com:8883"},
				Name:    "apac",
				Opts:    []mqttv5.Option{mqttv5.WithCredentials("apac-svc", []byte("token-2"))},
			},
		},
		mqttv5.WithGroupSharedOpts(
			mqttv5.WithClientID("fanout"),
			mqttv5.WithKeepAlive(30),
		),
		mqttv5.WithGroupPublishPolicy(mqttv5.GroupPublishBroadcast),
	)
	if err != nil {
		panic(err)
	}

	ctx := context.Background()
	if err := group.Connect(ctx); err != nil {
		panic(err)
	}
	defer group.Disconnect(ctx)

	msgs, _, err := group.Subscribe(ctx,
		[]mqttv5.TopicFilter{{Topic: "events/#", QoS: 1}})
	if err != nil {
		return
	}
	for m := range msgs {
		fmt.Println(m.Topic)
		_ = m.Ack()
	}
}

func (*ClientGroup) Connect 

func (g *ClientGroup) Connect(ctx context.Context) error

Connect dials every member. The default is parallel; use WithGroupSequentialLifecycle to serialise. The returned error is errors.Join of every member that failed — nil on full success. If at least one member succeeded the group is still usable for the healthy subset.

func (*ClientGroup) Disconnect 

func (g *ClientGroup) Disconnect(ctx context.Context) error

Disconnect tears down every member, which closes the merged channels and queues of the group's subscriptions. The default is parallel; use WithGroupSequentialLifecycle to serialise. Returned error is errors.Join of every member's Disconnect error. The group cannot be connected again.

func (*ClientGroup) Member 

func (g *ClientGroup) Member(name string) *Client

Member returns the Client for the named member, or nil when no such member exists.

func (*ClientGroup) Members 

func (g *ClientGroup) Members() []*Client

Members returns the underlying Clients in member order. Use for per-member operations the group API doesn't expose directly (Stats per member, manual Publish to a specific member, etc.).

func (*ClientGroup) Names 

func (g *ClientGroup) Names() []string

Names returns the member names in order.

func (*ClientGroup) Publish 

func (g *ClientGroup) Publish(ctx context.Context, opts PublishOptions) ([]GroupResult, error)

Publish dispatches opts across members per the configured GroupPublishPolicy and returns each member's part. Broadcast publishes to every member at once and fails with a *GroupError when fewer than WithGroupSuccess require succeed. RoundRobin and HashByTopic send the message once: they move to the next member only when one could not send it at all (not connected), so a broker refusal or ctx ending is returned as is and never duplicates the message.

func (*ClientGroup) PublishValue added in v0.12.0 

func (g *ClientGroup) PublishValue(ctx context.Context, opts PublishOptions, v any) ([]GroupResult, error)

PublishValue encodes v with each member's Encoding and publishes it as ClientGroup.Publish does; members with different Encodings send differently encoded copies. If any member has no Encoding it returns ErrNoEncoding and publishes nothing.

func (*ClientGroup) Subscribe 

func (g *ClientGroup) Subscribe(ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (<-chan *Message, map[string]SubscriptionToken, error)

Subscribe subscribes on every member and merges their messages into one channel; Ack on a message goes back to the member that delivered it. The channel's buffer and drop rules are those of Client.Subscribe, applied as each member delivers, with room for every member's QoS 1 and 2 messages. It closes once every member's subscription has ended: after ClientGroup.UnsubscribeAll, an Unsubscribe of each member, or Disconnect.

When fewer members than WithGroupSuccess requires subscribe — or ctx ends — the members that did are unsubscribed again and a *GroupError (or ctx's error) is returned. Otherwise the token map, keyed by member name, holds the members that subscribed; failures of the others are logged. Called before ClientGroup.Connect, it returns at once, as Client.Subscribe does, and each member's SUBSCRIBE goes out first on its connection.

func (*ClientGroup) SubscribeAs added in v0.12.0 

func (g *ClientGroup) SubscribeAs[T any](ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (<-chan *TypedMessage[T], map[string]SubscriptionToken, error)

SubscribeAs is ClientGroup.Subscribe with each payload decoded into a T by the delivering member's Encoding (or SubEncoding), as Client.SubscribeAs decodes it.

func (*ClientGroup) SubscribeQueue 

func (g *ClientGroup) SubscribeQueue(ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (*Queue[*Message], map[string]SubscriptionToken, error)

SubscribeQueue is the queue-merged variant of Subscribe, with the bound and drop policy of Client.SubscribeQueue applied to the merged queue as a whole.

func (*ClientGroup) SubscribeQueueAs added in v0.12.0 

func (g *ClientGroup) SubscribeQueueAs[T any](ctx context.Context, filters []TopicFilter, opts ...SubscribeOption) (*Queue[*TypedMessage[T]], map[string]SubscriptionToken, error)

SubscribeQueueAs is ClientGroup.SubscribeQueue with each payload decoded into a T, as ClientGroup.SubscribeAs decodes it.

func (*ClientGroup) Unsubscribe 

func (g *ClientGroup) Unsubscribe(ctx context.Context, name string, token SubscriptionToken) error

Unsubscribe tears down the named member's subscription.

func (*ClientGroup) UnsubscribeAll 

func (g *ClientGroup) UnsubscribeAll(ctx context.Context, tokens map[string]SubscriptionToken) error

UnsubscribeAll tears down every token in the supplied map. Errors are joined.

type ClientGroupOption 

type ClientGroupOption func(*clientGroupConfig)

ClientGroupOption configures a ClientGroup at construct time.

func WithGroupPublishPolicy 

func WithGroupPublishPolicy(p GroupPublishPolicy) ClientGroupOption

WithGroupPublishPolicy selects the ClientGroup.Publish dispatch policy. See GroupPublishPolicy.

func WithGroupSequentialLifecycle 

func WithGroupSequentialLifecycle() ClientGroupOption

WithGroupSequentialLifecycle disables the default parallel Connect / Disconnect / Subscribe across members. Useful in tests where deterministic ordering matters more than wall time.

func WithGroupSharedOpts 

func WithGroupSharedOpts(opts ...Option) ClientGroupOption

WithGroupSharedOpts applies opts to every member before that member's own GroupMember.Opts. Use for shared settings (KeepAlive, Logger, ReconnectBackoff). Per-member Opts override.

func WithGroupSuccess added in v0.11.0 

func WithGroupSuccess(s GroupSuccess) ClientGroupOption

WithGroupSuccess sets how many members must succeed for a broadcast Publish or a Subscribe to succeed. Default GroupSuccessAll.

type Config 

type Config struct {
	// BrokerURLs is the ordered list of broker URLs. The supervisor
	// rotates through it on reconnect; [Client.SetBrokers] replaces
	// it at runtime.
	BrokerURLs []string
	ClientID   string
	Username   string
	Password   []byte
	KeepAlive  uint16

	// CleanStart sets the CleanStart flag on the initial CONNECT
	// (§3.1.2.4). When not set explicitly it is true unless there is
	// session state to resume (see [WithStore]).
	CleanStart bool

	// SessionLossPolicy decides what happens to unacknowledged QoS 1/2
	// publishes when a connection starts without the old session.
	// Default [SessionLossRepublish].
	SessionLossPolicy SessionLossPolicy

	// OutboundTopicAliases replaces repeated QoS 0 topics with topic
	// aliases. Default false.
	OutboundTopicAliases bool

	// LenientDecoding tolerates the harmless broker violations listed at
	// [WithLenientDecoding] instead of disconnecting. Default false.
	LenientDecoding bool

	// OnResubscribeError fires when the broker refuses filters the client
	// re-subscribes after a session loss. Runs with the other lifecycle callbacks (see the package documentation).
	OnResubscribeError func(SubscriptionToken, error)

	// QoSDowngrade lowers a publish's QoS to the broker's Maximum QoS
	// instead of failing it with [ErrQoSNotSupported]. Default false.
	QoSDowngrade bool

	// OnServerRedirect fires when the broker redirects the client (§4.11).
	OnServerRedirect func(ServerRedirect)
	// FollowServerRedirects makes the client connect where a redirect
	// points.
	FollowServerRedirects bool

	// RetryInitialConnect makes Connect hand a failed first handshake
	// to the supervisor's reconnect loop instead of returning its error.
	RetryInitialConnect bool

	// CleanStartOnReconnect sets CleanStart on every CONNECT after
	// the first. Default false (preserve broker session for QoS 1/2
	// resume).
	CleanStartOnReconnect bool

	SessionExpiry  uint32
	ReceiveMaximum uint16

	// MaximumPacketSize caps the largest packet the broker may send
	// to this client (§3.1.2.11.4). Zero advertises no limit. A larger
	// packet closes the connection with DISCONNECT 0x95.
	MaximumPacketSize uint32

	// ReadBufferSize is the per-connection read window in bytes: how
	// much one read syscall can bring in. Default
	// [DefaultReadBufferSize].
	ReadBufferSize int

	// InboundTopicAliasMaximum is the inbound TopicAliasMaximum
	// (§3.1.2.11.5). Zero (default) tells the broker not to alias
	// inbound PUBLISHes. The outbound budget is taken automatically
	// from the broker's CONNACK.
	InboundTopicAliasMaximum uint16

	// RequestResponseInformation asks the broker for
	// ResponseInformation in CONNACK (§3.1.2.11.6).
	RequestResponseInformation bool

	// RequestProblemInformation asks the broker for ReasonString /
	// UserProperties on error responses (§3.1.2.11.7).
	RequestProblemInformation bool

	// ConnectUserProperties are sent verbatim as CONNECT user
	// properties.
	ConnectUserProperties []wire.UserProperty

	ConnectTimeout time.Duration
	PingTimeout    time.Duration

	// DisconnectFlushTimeout bounds the wait for a graceful
	// DISCONNECT write before the socket is forcibly closed.
	// Default 500ms.
	DisconnectFlushTimeout time.Duration

	WriteQueueSize int

	// WriteOverflowPolicy controls what QoS 0 Publish does when the
	// writer queue is full. WriteBlock (default) waits for room or
	// ctx; WriteDropNewest returns ErrWriteQueueFull immediately.
	// QoS 1/2 unaffected.
	WriteOverflowPolicy WriteOverflowPolicy

	// WriteBatchMax caps how many pre-encoded packets the writer
	// goroutine coalesces into one writev. 0 disables batching. See
	// [WithWriteBatch].
	WriteBatchMax int

	WillMessage *WillOptions

	// MaxSubscribeQueueSize caps each SubscribeQueue's length.
	// 0 means [DefaultMaxSubscribeQueueSize]; [UnboundedQueue] (any
	// negative value) removes the cap.
	MaxSubscribeQueueSize int

	// DropPolicy sets the default policy for full subscription
	// buffers. [Client.Subscribe] (chan) supports DropNewest only;
	// [Client.SubscribeQueue] honors both.
	DropPolicy DropPolicy

	// PublisherPoolSize > 0 enables a pool of that many publish-only
	// connections, which falls back to the main connection when no
	// member can send.
	PublisherPoolSize int

	// PublisherPoolRouting selects the pool member-picking strategy.
	PublisherPoolRouting PoolRoutingPolicy

	// PublisherPoolClientIDFn derives each pool member's ClientID
	// from the parent ClientID and a 1-based index. Default is
	// fmt.Sprintf("%s-pub-%d", parent, idx), or empty (broker-assigned)
	// when the parent's is empty.
	PublisherPoolClientIDFn func(parent string, idx int) string

	// PublishMode selects QoS 0 write-completion semantics. QoS 1/2
	// always wait for the broker's ack.
	PublishMode PublishMode

	// Encoding encodes and decodes the payloads of the typed calls
	// (PublishValue, SubscribeAs and their variants). See [WithEncoding].
	Encoding Encoding

	// OnConnectionUp fires on each successful CONNECT/CONNACK with
	// what the broker granted, possibly after Connect has returned.
	// Runs with the other lifecycle callbacks (see the package documentation).
	OnConnectionUp func(ConnackInfo)

	// OnConnectionDown fires on unexpected connection loss (not on
	// user-initiated Disconnect). Return false to stop the
	// supervisor, which waits for the answer; a subsequent Connect
	// restarts it. Runs with the other lifecycle callbacks (see the package documentation).
	OnConnectionDown func() bool

	// OnConnectError fires per failed CONNECT attempt (dial failure,
	// CONNACK refusal, AUTH-loop error). Observability only; the
	// supervisor retries regardless. Runs with the other lifecycle callbacks (see the package documentation).
	OnConnectError func(err error)

	// OnReconnectAttempt fires before each reconnect dial (not the
	// initial Connect), which waits for it. attempt starts at 1.
	// Runs with the other lifecycle callbacks (see the package documentation).
	OnReconnectAttempt func(attempt int, brokerURL string)

	// StatsEnabled toggles the [Client.Stats] counter set. Default
	// false — when disabled the hot path skips every atomic
	// increment and Stats returns the zero value.
	StatsEnabled bool

	// OnServerDisconnect fires when the broker sends a DISCONNECT
	// (§3.14).
	// Fires after the connection is marked down and before
	// OnConnectionDown. May call [Client.SetBrokers] to honour a
	// ServerMoved / UseAnotherServer redirect on the next attempt.
	// Runs with the other lifecycle callbacks (see the package documentation).
	OnServerDisconnect func(DisconnectInfo)

	// Authenticator drives MQTT v5 enhanced authentication when set.
	Authenticator Authenticator

	// OnReauthenticated fires when a re-authentication (§4.12) the
	// client started with Reauthenticate concludes successfully (broker
	// AUTH 0x00 Success), possibly after Reauthenticate has returned.
	// Observability only: the success is also Reauthenticate's return
	// value. Runs with the other lifecycle callbacks (see the package documentation).
	OnReauthenticated func()

	// ConnectPacketBuilder is invoked immediately before each CONNECT
	// is serialised and may change its credentials and user
	// properties — typically rotating Username/Password per attempt
	// for OAuth refresh. ctx is the per-attempt context bounded by
	// ConnectTimeout. A non-nil error fails the attempt; the
	// supervisor retries after backoff and fires OnConnectError.
	ConnectPacketBuilder func(ctx context.Context, opts *ConnectOptions) error

	// ReconnectBackoff returns the delay before retry attempt N.
	// Defaults to [DefaultReconnectBackoff].
	ReconnectBackoff Backoff

	Logger *slog.Logger

	// Store persists session state across process restarts. Nil (the
	// default) keeps it in memory: in-flight QoS 1/2 messages survive
	// reconnects but not a restart.
	Store session.Store

	// OnStoreFailure fires once when a Store write fails and the client
	// has stopped because of it (see [WithStore]). Runs with the other lifecycle callbacks (see the package documentation).
	OnStoreFailure func(err error)

	TLSConfig *tls.Config
	Dialer    *net.Dialer

	// DialFunc replaces the built-in TCP/TLS dial path. Takes
	// precedence over TLSConfig and Dialer. Use for SOCKS / mTLS /
	// WebSocket (see transport/ws) / in-memory test transports.
	// When set, broker URL scheme validation is skipped.
	DialFunc transport.DialFunc
	// contains filtered or unexported fields
}

Config carries the resolved configuration for a Client. Construct via New.

type ConnackInfo added in v0.11.0 

type ConnackInfo struct {
	// Broker is the URL of the broker the connection reached: with
	// [WithBrokers], the one the client failed over to.
	Broker string

	// SessionPresent reports whether the broker resumed an existing
	// session.
	SessionPresent bool

	// SessionExpiry is the effective Session Expiry Interval in seconds:
	// the broker's value when it overrode the requested one.
	SessionExpiry uint32

	// KeepAlive is the effective keep-alive in seconds: the broker's
	// Server Keep Alive when sent, otherwise the requested value.
	// Zero disables keep-alive.
	KeepAlive uint16

	// ReceiveMaximum caps the QoS 1/2 PUBLISHes the client may have
	// unacknowledged (default 65,535).
	ReceiveMaximum uint16

	// MaximumQoS is the highest QoS the broker accepts (default 2).
	MaximumQoS byte

	// RetainAvailable reports whether retained messages are accepted
	// (default true).
	RetainAvailable bool

	// MaximumPacketSize is the largest packet the broker accepts, in
	// bytes; zero means no limit beyond the protocol's.
	MaximumPacketSize uint32

	// TopicAliasMaximum is the number of outbound topic aliases the
	// broker accepts (default 0: none).
	TopicAliasMaximum uint16

	// AssignedClientID is the identifier the broker chose when the
	// client connected without one.
	AssignedClientID string

	// Subscription features (default true when not sent).
	WildcardSubscriptionAvailable    bool
	SubscriptionIdentifiersAvailable bool
	SharedSubscriptionAvailable      bool

	ResponseInformation  string
	ServerReference      string
	ReasonString         string
	UserProperties       []wire.UserProperty
	AuthenticationMethod string
	AuthenticationData   []byte
}

ConnackInfo is what the broker granted in CONNACK (§3.2.2.3), with the spec's defaults applied to properties it did not send. The client enforces these limits on everything it sends: a publish that exceeds them fails before it reaches the wire.

type ConnectOptions added in v0.11.0 

type ConnectOptions struct {
	Username       string
	Password       []byte
	UserProperties []UserProperty
}

ConnectOptions are the parts of a CONNECT that WithConnectPacketBuilder may change before each attempt, typically to present a fresh credential.

type ContextAuthenticator added in v0.11.0 

type ContextAuthenticator interface {
	ContinueContext(ctx context.Context, brokerData []byte) (response []byte, done bool, err error)
}

ContextAuthenticator is an optional interface an Authenticator may implement so answering a broker challenge can be cancelled. The client then calls ContinueContext instead of Continue, with a ctx bounded by the connect timeout. Mechanisms that do I/O to answer (a token service, a hardware key) should implement it.

type DisconnectInfo added in v0.11.0 

type DisconnectInfo struct {
	ReasonCode   ReasonCode
	ReasonString string
	// ServerReference names another server to use, with reason 0x9C
	// Use another server or 0x9D Server moved.
	ServerReference string
	UserProperties  []UserProperty
}

DisconnectInfo is a DISCONNECT the broker sent.

type DisconnectOptions added in v0.11.0 

type DisconnectOptions struct {
	ReasonCode ReasonCode // 0x00 Normal disconnection, 0x04 with Will, or an error code
	// SessionExpiryInterval replaces the value CONNECT set; it cannot
	// be non-zero if CONNECT sent 0 [MQTT-3.14.2-2].
	SessionExpiryInterval *uint32
	ReasonString          string
	UserProperties        []UserProperty
}

DisconnectOptions is the DISCONNECT Client.DisconnectWith sends.

type DropPolicy 

type DropPolicy uint8

DropPolicy controls behaviour when a bounded subscription buffer fills. 

const (
	// DropNewest discards the inbound message + acks it so the
	// broker doesn't retry.
	DropNewest DropPolicy = iota
	// DropOldest evicts the buffer head to admit the inbound
	// message. SubscribeQueue only — chan-based [Client.Subscribe]
	// rejects this with [ErrChanDropOldestUnsupported].
	DropOldest
)

func (DropPolicy) String 

func (p DropPolicy) String() string

String returns a stable debug name for the policy.

type Encoding added in v0.12.0 

type Encoding interface {
	// ContentType is the MQTT 5 Content Type a typed publish declares,
	// such as "application/json". Empty declares none.
	ContentType() string

	// Marshal encodes v.
	Marshal(v any) ([]byte, error)

	// Unmarshal decodes data into v, a non-nil pointer. It must not
	// keep data, or slices of it, once it returns: on a [SubZeroCopy]
	// subscription data is the network frame, reused after the message
	// is acked.
	Unmarshal(data []byte, v any) error
}

Encoding turns Go values into payloads and back for the typed calls: Client.PublishValue and Client.SubscribeAs with their queue, callback, ClientGroup and QueuePublisher variants. One Encoding serves every payload type, so a client sets it once (WithEncoding) and names the type per call. The codec/json and codec/msgpack submodules provide one each. Implementations must be safe for concurrent use.

type GroupError added in v0.11.0 

type GroupError struct {
	Op      string // "publish" or "subscribe"
	Need    int
	Results []GroupResult
}

GroupError reports a group operation that fewer members completed than WithGroupSuccess requires. errors.Is and errors.As see every member's error.

func (*GroupError) Error added in v0.11.0 

func (e *GroupError) Error() string

func (*GroupError) Unwrap added in v0.11.0 

func (e *GroupError) Unwrap() []error

type GroupMember 

type GroupMember struct {
	// Brokers are the URLs this member dials: one, or several the
	// member fails over between as a [Client] with [WithBrokers] does.
	// Required; neither Opts nor [WithGroupSharedOpts] may set brokers.
	Brokers []string

	// Name is an optional identifier surfaced in logs, returned in
	// the Subscribe token map, and accepted by ClientGroup.Member.
	// Defaults to "member-N" (1-based) when empty.
	Name string

	// Opts are applied AFTER any WithGroupSharedOpts shared options.
	// Use to override or extend the shared opts per-broker (auth,
	// TLS, ClientID, per-member callbacks, etc.).
	Opts []Option
}

GroupMember describes one member of a ClientGroup. Each member becomes a full Client with its own session, supervisor, and CONNECT identity.

Use Opts for per-member settings — WithCredentials, WithTLSConfig, WithAuthenticator, WithClientID, WithConnectPacketBuilder, a per-member WithOnConnectionUp closing over Name for identity, etc. Opts are applied AFTER WithGroupSharedOpts so per-member settings win.

type GroupPublishPolicy 

type GroupPublishPolicy uint8

GroupPublishPolicy selects how ClientGroup.Publish dispatches across members. For HA failover across interchangeable brokers use WithBrokers on a single Client instead — ClientGroup keeps N parallel sessions where every member is treated as itself. 

const (
	// GroupPublishBroadcast (default) publishes to every member at
	// once. Use when members carry different data (bridge / mirror
	// semantic). [WithGroupSuccess] decides how many must succeed.
	GroupPublishBroadcast GroupPublishPolicy = iota

	// GroupPublishRoundRobin spreads publishes over the group's
	// brokers, one member per call. The per-call starting member
	// advances by one; a member that cannot send (not connected) passes
	// the message to the next. Use when any of the brokers will do for
	// a message because its consumers subscribe to all of them. For
	// publish throughput into one broker or cluster use
	// [WithPublisherPool] instead.
	GroupPublishRoundRobin

	// GroupPublishHashByTopic picks the member by FNV-1a of
	// opts.Topic, preserving per-topic ordering across the group while
	// every member is connected; a member that cannot send passes the
	// message to the next, briefly reordering that topic. Empty Topic
	// collapses to member 0.
	GroupPublishHashByTopic
)

func (GroupPublishPolicy) String 

func (p GroupPublishPolicy) String() string

String returns a stable debug name for the policy.

type GroupResult added in v0.11.0 

type GroupResult struct {
	Member string
	Err    error // nil: the member succeeded
}

GroupResult is one member's part in a group operation.

type GroupSuccess added in v0.11.0 

type GroupSuccess int

GroupSuccess is how many members must succeed for a broadcast ClientGroup.Publish or a ClientGroup.Subscribe to succeed. 

const (
	// GroupSuccessAll (default) needs every member.
	GroupSuccessAll GroupSuccess = 0
	// GroupSuccessAny needs one member.
	GroupSuccessAny GroupSuccess = 1
)

func GroupSuccessQuorum added in v0.11.0 

func GroupSuccessQuorum(n int) GroupSuccess

GroupSuccessQuorum needs n members.

type HandlerFunc 

type HandlerFunc func(*Message)

HandlerFunc runs synchronously on the read goroutine for each matching inbound PUBLISH. Keep it short: while it runs the connection reads nothing, so other subscriptions' messages and the acknowledgements of the client's own publishes wait behind it. The keep-alive does not take that wait for a dead broker. Use Subscribe / SubscribeQueue for work that may take long.

type MemoryPublisherQueue 

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

MemoryPublisherQueue is an in-memory PublisherQueue: entries do not survive the process. Use the queue/file submodule to keep them across restarts.

func NewMemoryPublisherQueue 

func NewMemoryPublisherQueue() *MemoryPublisherQueue

NewMemoryPublisherQueue returns an empty in-memory queue.

func (*MemoryPublisherQueue) Ack 

func (q *MemoryPublisherQueue) Ack(_ context.Context, seq uint64) error

Ack implements PublisherQueue.

func (*MemoryPublisherQueue) Close 

func (q *MemoryPublisherQueue) Close() error

Close implements PublisherQueue.

func (*MemoryPublisherQueue) Enqueue 

func (q *MemoryPublisherQueue) Enqueue(_ context.Context, e QueueEntry, limit QueueLimit) (uint64, []QueueEntry, error)

Enqueue implements PublisherQueue.

func (*MemoryPublisherQueue) Len 

func (q *MemoryPublisherQueue) Len(context.Context) (int, error)

Len implements PublisherQueue.

func (*MemoryPublisherQueue) Peek added in v0.11.0 

func (q *MemoryPublisherQueue) Peek(_ context.Context, after uint64, n int) ([]QueueEntry, error)

Peek implements PublisherQueue.

type Message 

type Message struct {
	Topic      string
	Payload    []byte
	Properties Properties
	QoS        byte
	Retain     bool
	Dup        bool
	PacketID   uint16
	// contains filtered or unexported fields
}

Message is one inbound PUBLISH as seen by one subscription handler.

By default a Message owns its memory: Topic, Payload and Properties are copied out of the network frame before any handler runs, stay valid for as long as the Message is reachable, and may be retained freely. When a PUBLISH matches several subscriptions, each handler receives its own *Message; the handles share the same read-only bytes.

Subscriptions created with SubZeroCopy instead receive Messages whose Topic, Payload and Properties alias the pooled network frame. Those fields are valid only until every handler's Ack has returned.

Ack acknowledges the message to the broker (PUBACK for QoS 1, PUBREC for QoS 2, emitted in arrival order per §4.6) once every handler that received the PUBLISH has acked. Ack is idempotent per Message: later calls are no-ops and never affect any other message.

func (*Message) Ack 

func (m *Message) Ack() error

Ack acknowledges the message for this handler. It is safe to call more than once and from any goroutine; only the first call counts. For QoS 0 messages Ack is only needed on zero-copy subscriptions, where it returns the network frame to the pool.

func (*Message) ClonePayload 

func (m *Message) ClonePayload() []byte

ClonePayload returns a copy of Payload. Only needed on zero-copy subscriptions; owned messages may simply keep m.Payload.

func (*Message) CloneTopic 

func (m *Message) CloneTopic() string

CloneTopic returns a copy of Topic. Only needed on zero-copy subscriptions; owned messages may simply keep m.Topic.

func (*Message) PublishOptions added in v0.11.0 

func (m *Message) PublishOptions() PublishOptions

PublishOptions returns options that publish m again as it arrived: topic, payload, QoS, Retain, and the properties a client may send — Payload Format Indicator, the Message Expiry Interval the broker left it, Content Type, Response Topic, Correlation Data and User Properties. A bridge changes the Topic or QoS and publishes it on another connection. Retain is the flag as delivered: the publisher's only on a subscription with RetainAsPublished. The options share m's memory, so on a zero-copy subscription publish them before acking m.

type Option 

type Option func(*Config) error

Option mutates a Config during New; a non-nil return aborts construction.

func WithAuthenticator 

func WithAuthenticator(a Authenticator) Option

WithAuthenticator enables MQTT v5 enhanced authentication. When combined with WithPublisherPool, the same Authenticator runs concurrently across the parent and every pool member; implementations used that way must be safe for concurrent use.

func WithBroker 

func WithBroker(url string) Option

WithBroker sets a single broker URL. Supported schemes: mqtt://, tcp://, mqtts://, tls://, ssl://; default ports 1883 / 8883 fill in when missing. For ws:// or wss:// pair with WithDialFunc.

func WithBrokers 

func WithBrokers(urls ...string) Option

WithBrokers sets the broker URL list for failover. Each client starts at a random entry, so a fleet spreads across the list. Client.Connect tries the next entry when one fails; each reconnect attempt takes the next entry, so the first one leaves the broker that just dropped. The client stays on whichever broker accepts it. Client.SetBrokers replaces the list at runtime.

Brokers that do not share session state answer a failover CONNECT with Session Present = 0: unacknowledged QoS 1/2 publishes then follow WithSessionLossPolicy and subscriptions are re-issued.

func WithCleanStart 

func WithCleanStart(b bool) Option

WithCleanStart sets the CleanStart flag on the initial CONNECT. Without it the first CONNECT starts clean unless there is session state to resume — state left by an earlier Connect on this Client or loaded from WithStore — in which case it resumes (CleanStart=0). WithCleanStart(true) always starts a new session and discards that state. Reconnects use WithCleanStartOnReconnect.

func WithCleanStartOnReconnect 

func WithCleanStartOnReconnect(b bool) Option

WithCleanStartOnReconnect sets CleanStart on every CONNECT after the first. Default false — preserve broker session for QoS 1/2 resume.

func WithClientID 

func WithClientID(id string) Option

WithClientID sets the MQTT ClientID. Empty string asks the broker to assign one via the AssignedClientIdentifier CONNACK property.

func WithConnectPacketBuilder 

func WithConnectPacketBuilder(fn func(ctx context.Context, opts *ConnectOptions) error) Option

WithConnectPacketBuilder lets fn change the CONNECT's credentials and user properties immediately before each attempt — canonical OAuth token-refresh hook. ctx is the per-attempt context bounded by ConnectTimeout. A non-nil error fails the attempt; the supervisor retries after backoff and fires OnConnectError.

func WithConnectTimeout 

func WithConnectTimeout(d time.Duration) Option

WithConnectTimeout caps the CONNECT handshake (dial + CONNECT/CONNACK).

func WithConnectUserProperties 

func WithConnectUserProperties(p []wire.UserProperty) Option

WithConnectUserProperties replaces the CONNECT user-property list. The slice is referenced as-is; do not mutate after passing.

func WithConnectUserProperty 

func WithConnectUserProperty(key, value string) Option

WithConnectUserProperty appends one CONNECT user property. May be called multiple times; for bulk replacement see WithConnectUserProperties.

func WithCredentials 

func WithCredentials(user string, pass []byte) Option

WithCredentials sets username and password sent in the CONNECT packet.

func WithDialFunc 

func WithDialFunc(fn transport.DialFunc) Option

WithDialFunc replaces the built-in TCP/TLS dial path. Takes precedence over WithDialer / WithTLSConfig. Use for SOCKS / HTTP proxies, per-dial mTLS, WebSocket (transport/ws), or in-memory test transports. ctx is bounded by ConnectTimeout. When set, broker URL scheme validation is skipped. Nil rejected.

The transport/ws sibling submodule exposes a [ws.DialFunc] helper that bakes in DialOpts (TLS config, custom headers, etc.) and returns a value directly usable here: 

mqttv5.WithDialFunc(ws.DialFunc(ws.DialOpts{TLSConfig: tlsCfg}))

func WithDialer 

func WithDialer(d *net.Dialer) Option

WithDialer overrides the default *net.Dialer.

func WithDisconnectFlushTimeout 

func WithDisconnectFlushTimeout(d time.Duration) Option

WithDisconnectFlushTimeout caps the wait for a graceful DISCONNECT write before the socket is forcibly closed. Default DefaultDisconnectFlushTimeout.

func WithDropPolicy 

func WithDropPolicy(p DropPolicy) Option

WithDropPolicy sets the default per-subscription drop policy. Per-call override via SubDropPolicy; see DropPolicy.

func WithEncoding added in v0.12.0 

func WithEncoding(e Encoding) Option

WithEncoding sets the Encoding of the client's typed calls. A subscription can use another with SubEncoding.

func WithFollowServerRedirects added in v0.11.0 

func WithFollowServerRedirects() Option

WithFollowServerRedirects makes the client connect where the broker redirects it: after 0x9D Server moved the reference replaces the broker list (Client.SetBrokers); after 0x9C Use another server only the next connection attempt uses it. A reference that is a host or host:port keeps the current URL's scheme and path. Off by default: a redirect is reported (WithOnServerRedirect) and otherwise ignored.

func WithInboundTopicAliasMaximum 

func WithInboundTopicAliasMaximum(n uint16) Option

WithInboundTopicAliasMaximum advertises how many inbound topic aliases this client accepts (§3.1.2.11.5). Zero (default) disables inbound aliasing.

func WithKeepAlive 

func WithKeepAlive(s uint16) Option

WithKeepAlive sets the keepalive interval (seconds, range 1..65535). Pass 0 to WithoutKeepAlive instead.

func WithLenientDecoding added in v0.11.0 

func WithLenientDecoding() Option

WithLenientDecoding tolerates two harmless protocol violations seen from real brokers — a non-minimally encoded Remaining Length and reserved flag bits on PINGRESP — logging each at Warn instead of disconnecting. Every other violation still closes the connection with the spec's reason code (0x81 Malformed Packet, 0x82 Protocol Error, or a more specific one) and counts in Stats.ProtocolErrors.

func WithLogger 

func WithLogger(l *slog.Logger) Option

WithLogger sets the *slog.Logger. Default slog.Default.

func WithMaxSubscribeQueueSize 

func WithMaxSubscribeQueueSize(n int) Option

WithMaxSubscribeQueueSize caps each Client.SubscribeQueue's length. Default DefaultMaxSubscribeQueueSize; UnboundedQueue removes the cap. When a queue is full its DropPolicy applies. Per-call override via SubMaxQueueSize.

func WithMaximumPacketSize 

func WithMaximumPacketSize(n uint32) Option

WithMaximumPacketSize caps the largest packet the broker may send to this client (§3.1.2.11.4). Zero advertises no limit. A broker that sends a larger packet is disconnected with reason 0x95.

func WithOnConnectError 

func WithOnConnectError(fn func(error)) Option

WithOnConnectError fires per failed CONNECT attempt (dial err, CONNACK refusal, AUTH-loop err). Observability only — the supervisor retries regardless. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnConnectionDown 

func WithOnConnectionDown(fn func() bool) Option

WithOnConnectionDown registers a callback fired when the connection is lost. Does NOT fire on user-initiated Disconnect. Return false to terminate the supervisor — no further reconnect attempts; the supervisor waits for the answer. A subsequent Connect on the same Client re-starts the lifecycle. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnConnectionUp 

func WithOnConnectionUp(fn func(ConnackInfo)) Option

WithOnConnectionUp fires on each successful CONNECT/CONNACK with what the broker granted (session present, limits, assigned ClientID, effective keep-alive, ...); Client.ServerInfo returns the same for the current connection. It may run after Client.Connect has returned. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnReauthenticated added in v0.10.0 

func WithOnReauthenticated(fn func()) Option

WithOnReauthenticated registers a callback fired when a re-authentication (§4.12) the client started with Reauthenticate concludes successfully (broker AUTH 0x00 Success), possibly after Reauthenticate has returned. Observability only. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnReconnectAttempt 

func WithOnReconnectAttempt(fn func(attempt int, brokerURL string)) Option

WithOnReconnectAttempt fires before each reconnect dial (not the initial Client.Connect), which waits for it. attempt starts at 1 after a connection that lasted, and keeps counting across connections that dropped sooner than the reconnect delay before them (see WithReconnectBackoff). Runs with the other lifecycle callbacks (see the package documentation).

func WithOnResubscribeError added in v0.11.0 

func WithOnResubscribeError(fn func(SubscriptionToken, error)) Option

WithOnResubscribeError observes filters the broker refuses when the client re-subscribes after a session loss (Session Present = 0). A subscription that loses every filter closes, and its token's Err reports why; one that keeps some carries on with those. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnServerDisconnect 

func WithOnServerDisconnect(fn func(DisconnectInfo)) Option

WithOnServerDisconnect fires when the broker sends a DISCONNECT (§3.14), with its reason, reason string, server reference and user properties. Fires after the connection is marked down and before OnConnectionDown; may call Client.SetBrokers to honour a ServerMoved / UseAnotherServer redirect on the next attempt, which waits for it. Not fired for socket-level errors or client-initiated Disconnect. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnServerRedirect added in v0.11.0 

func WithOnServerRedirect(fn func(ServerRedirect)) Option

WithOnServerRedirect calls fn when the broker refuses a CONNECT, or ends the connection, with reason 0x9C Use another server or 0x9D Server moved and a Server Reference (§4.11). May call Client.SetBrokers: the next attempt waits for it. Runs with the other lifecycle callbacks (see the package documentation).

func WithOnStoreFailure added in v0.11.0 

func WithOnStoreFailure(fn func(err error)) Option

WithOnStoreFailure registers a callback fired once when a session store write failed and the client has stopped because of it (see WithStore); err is a *StoreError. Call Client.Connect, from the callback or later, to start again from what the store holds. Runs with the other lifecycle callbacks (see the package documentation).

func WithOutboundTopicAliases added in v0.11.0 

func WithOutboundTopicAliases() Option

WithOutboundTopicAliases makes QoS 0 publishes replace a repeated topic with a topic alias (§3.3.2.3.4) while the broker's Topic Alias Maximum lasts: the first publish to a topic registers the alias, later ones send only the alias. It saves bandwidth for long topics. QoS 1/2 publishes always carry their topic, since they may be resent on a connection where the alias is unknown. Off by default.

func WithPingTimeout 

func WithPingTimeout(d time.Duration) Option

WithPingTimeout sets how long the client waits for any packet after a PINGREQ before declaring the connection dead and reconnecting. It must be shorter than the keep-alive (New rejects it otherwise); when the broker grants a shorter keep-alive than requested, half the granted keep-alive is used instead. Default: the smaller of DefaultPingTimeout and half the keep-alive. Raise it only on links whose round trip exceeds a few seconds (satellite, congested cellular).

func WithPublishMode 

func WithPublishMode(m PublishMode) Option

WithPublishMode selects QoS 0 Client.Publish semantics. See PublishMode. QoS 1/2 unaffected.

func WithPublisherPool 

func WithPublisherPool(size int) Option

WithPublisherPool opens size publish-only connections, beside the main one, to the client's own brokers; Client.Publish spreads over them, and falls back to the main connection when none can send. 0 (the default) disables the pool. Each member connects with Session Expiry 0 (and Clean Start the first time), so its session ends with its connection: a QoS 1/2 publish interrupted by a member's reconnect follows the session-loss policy (sent again, possibly duplicated). Subscriptions, the Will, the session store and QueuePublisher stay on the main connection; members read their own acknowledgements, so a slow Client.SubscribeCallback handler does not delay pooled publishes. The broker sees each member as a client of its own (see WithPublisherPoolClientIDFn), so No Local does not hide a pooled publish from the main connection's subscriptions.

Example

ExampleWithPublisherPool dedicates N publish-only connections to the same broker, lifting publish throughput above what one keep-alive socket can absorb. Pair with mqttv5.PoolRoutingHashByTopic to preserve per-topic ordering. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, err := mqttv5.New(
		mqttv5.WithBroker("mqtt://localhost:1883"),
		mqttv5.WithPublisherPool(4),
		mqttv5.WithPublisherPoolRouting(mqttv5.PoolRoutingHashByTopic),
	)
	if err != nil {
		panic(err)
	}

	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	// Round-robin / hashed across the 4 pool members under the hood.
	for i := range 100 {
		_ = cli.Publish(ctx, mqttv5.PublishOptions{
			Topic:   fmt.Sprintf("events/tick/%d", i%4),
			Payload: fmt.Appendf(nil, `{"n":%d}`, i),
			QoS:     1,
		})
	}
}

func WithPublisherPoolClientIDFn 

func WithPublisherPoolClientIDFn(fn func(parent string, idx int) string) Option

WithPublisherPoolClientIDFn derives each pool member's ClientID from the parent's configured ClientID and a 1-based index. The default is "<parent>-pub-<n>" (DefaultPublisherPoolClientIDFormat), or, for a parent without a ClientID, empty, so the broker assigns each member's. Broker ACLs or policies keyed on the ClientID must admit these IDs, and fn must give IDs no other client uses.

func WithPublisherPoolRouting 

func WithPublisherPoolRouting(p PoolRoutingPolicy) Option

WithPublisherPoolRouting selects the publisher-pool member-picking strategy. See PoolRoutingPolicy. No-op without a pool.

func WithQoSDowngrade added in v0.11.0 

func WithQoSDowngrade() Option

WithQoSDowngrade sends a publish whose QoS is above the broker's Maximum QoS (CONNACK, §3.2.2.3.4) at that maximum instead of failing it with ErrQoSNotSupported. It weakens the delivery guarantee the caller asked for — QoS 2 becomes at-least-once — so it is off by default. Brokers such as AWS IoT Core advertise Maximum QoS 1.

func WithReadBufferSize added in v0.11.0 

func WithReadBufferSize(n int) Option

WithReadBufferSize sets the per-connection read window in bytes: how much one read syscall can bring in. A larger window cuts syscalls for streams of small messages at the cost of that much memory per connection. Default DefaultReadBufferSize.

func WithReceiveMaximum 

func WithReceiveMaximum(n uint16) Option

WithReceiveMaximum sets the Receive Maximum (§3.1.2.11.3): how many QoS 1/2 messages the broker may send that the client has not yet acknowledged. A delivered message is acknowledged once the application acks it, so this is also how many QoS 1/2 messages a subscription whose consumer falls behind can hold beyond its buffer, and what that costs in memory (see Client.Subscribe). A broker that sends more is disconnected with reason 0x93. Default DefaultReceiveMaximum; 0 keeps the default.

func WithReconnectBackoff 

func WithReconnectBackoff(b Backoff) Option

WithReconnectBackoff sets the supervisor's reconnect backoff: the delay before each reconnect attempt, by attempt number. The count starts over only once a connection has outlived the delay before it, so a broker that drops connections right after accepting them is retried at the growing interval rather than at the first one. See ConstantBackoff and ExponentialBackoff; default DefaultReconnectBackoff.

func WithRequestProblemInformation 

func WithRequestProblemInformation(b bool) Option

WithRequestProblemInformation asks the broker for ReasonString / UserProperties on error responses (§3.1.2.11.7). Default true — debugging a CONNECT or PUBLISH refusal without these is "why was this rejected?" with no answer; the wire cost is a handful of bytes on the error path only. Pass false to opt out: the CONNECT then carries Request Problem Information = 0.

func WithRequestResponseInformation 

func WithRequestResponseInformation(b bool) Option

WithRequestResponseInformation asks the broker for ResponseInformation in CONNACK (§3.1.2.11.6) — used by the request/response pattern (§4.10).

func WithRetryInitialConnect added in v0.11.0 

func WithRetryInitialConnect() Option

WithRetryInitialConnect lets the client start while the broker is unreachable: when the first CONNECT fails, Connect reports the error to WithOnConnectError, returns nil, and the supervisor retries with WithReconnectBackoff as after a lost connection. Publish and Subscribe return ErrNotConnected until then (QueuePublisher queues); Client.AwaitConnection waits for the connection. Without it, Connect returns the failure: with WithBrokers, once every broker has failed.

func WithSessionExpiry 

func WithSessionExpiry(s uint32) Option

WithSessionExpiry sets the Session Expiry Interval (§3.1.2.11.2) in seconds. Default DefaultSessionExpiry (300 s / 5 min) — enough for QoS 1/2 resume across a typical reconnect blip. Zero ends the session with the network connection (no broker-side retention).

func WithSessionLossPolicy added in v0.11.0 

func WithSessionLossPolicy(p SessionLossPolicy) Option

WithSessionLossPolicy sets the SessionLossPolicy. Default SessionLossRepublish.

func WithStats 

func WithStats() Option

WithStats enables the Client.Stats counter set. When disabled the hot path skips every atomic increment and Stats returns the zero value. Off by default; enable to scrape counters into Prometheus / OpenTelemetry / expvar.

func WithStore 

func WithStore(s session.Store) Option

WithStore persists session state in s so a restarted process resumes its QoS 1/2 flows: unacknowledged publishes are resent, QoS 2 exchanges continue at the phase they reached, and a broker-assigned ClientID is reused. Each state change is written to s before the packet that depends on it is sent (see package session). Without a store the same state is kept in memory and survives reconnects only.

A failed write of a new message's record fails that Publish. Any other failed write means the store no longer holds what the session has done, so the client stops rather than go on without the guarantee the store exists for: it sends DISCONNECT 0x80 (the broker publishes the Will), ends as Client.Disconnect would, and fires WithOnStoreFailure. QoS 1/2 Publish calls still waiting return the failure, a *StoreError matching ErrStoreFailed; their messages stay in the store. The next Client.Connect reloads the session from s, as a restarted process would.

The client does not close s. Use the store/file submodule for a file-backed store, or session.MemoryStore in tests.

func WithTLSConfig 

func WithTLSConfig(t *tls.Config) Option

WithTLSConfig supplies a *tls.Config for mqtts:// / tls:// dials. Ignored for plain TCP schemes.

func WithWill 

func WithWill(w *WillOptions) Option

WithWill attaches a will message to be published by the broker if this client disconnects ungracefully.

func WithWriteBatch 

func WithWriteBatch(n int) Option

WithWriteBatch coalesces up to n queued packets into one writev syscall. Off by default (n=0 or 1). It saves syscalls when many goroutines publish concurrently and packets queue behind the writer; with one publisher or large payloads there is little to coalesce. Measure with your workload before enabling it.

func WithWriteOverflowPolicy 

func WithWriteOverflowPolicy(p WriteOverflowPolicy) Option

WithWriteOverflowPolicy selects the QoS 0 writer-queue overflow policy. Only meaningful for PublishFireAndForget — QoS 1/2 and PublishWaitForFlush always wait for either the broker or the writer to be ready.

func WithWriteQueueSize 

func WithWriteQueueSize(n int) Option

WithWriteQueueSize sets the internal MPSC write-channel buffer. Larger absorbs publish bursts; costs memory and queue latency.

func WithoutKeepAlive 

func WithoutKeepAlive() Option

WithoutKeepAlive disables MQTT keepalive entirely (no PINGREQ goroutine). Rarely correct in production — without PINGREQ the broker cannot detect a half-open connection from this side.

type PacketType added in v0.11.0 

type PacketType = wire.PacketType

PacketType is an MQTT control packet type.

type PoolRoutingPolicy 

type PoolRoutingPolicy uint8

PoolRoutingPolicy selects how the publisher pool picks the starting member per Client.Publish. Unhealthy starting members fall through to the next member regardless of policy. 

const (
	// PoolRoutingRoundRobin (default) cycles through members; same
	// topic may arrive at the broker out of order.
	PoolRoutingRoundRobin PoolRoutingPolicy = iota

	// PoolRoutingHashByTopic picks the starting member by FNV-1a of
	// opts.Topic, preserving per-topic ordering while the pool is
	// healthy. Empty Topic collapses to member 0.
	PoolRoutingHashByTopic
)

func (PoolRoutingPolicy) String 

func (p PoolRoutingPolicy) String() string

String returns a stable debug name for the policy.

type Properties added in v0.11.0 

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

Properties are the properties of a received PUBLISH (§3.3.2.3). Their memory follows the Message's: owned unless the subscription is zero-copy.

func (Properties) ContentType added in v0.11.0 

func (p Properties) ContentType() string

ContentType returns the Content Type, or "".

func (Properties) CorrelationData added in v0.11.0 

func (p Properties) CorrelationData() []byte

CorrelationData returns the Correlation Data, or nil.

func (Properties) MessageExpiryInterval added in v0.11.0 

func (p Properties) MessageExpiryInterval() (uint32, bool)

MessageExpiryInterval returns the lifetime in seconds the broker left the message.

func (Properties) PayloadFormatIndicator added in v0.11.0 

func (p Properties) PayloadFormatIndicator() (byte, bool)

PayloadFormatIndicator returns the Payload Format Indicator: 1 for UTF-8 text, 0 for bytes.

func (Properties) ResponseTopic added in v0.11.0 

func (p Properties) ResponseTopic() string

ResponseTopic returns the Response Topic, or "".

func (Properties) SubscriptionIdentifiers added in v0.11.0 

func (p Properties) SubscriptionIdentifiers() iter.Seq[uint32]

SubscriptionIdentifiers yields the identifiers of the subscriptions the broker matched (§3.3.4).

func (Properties) UserProperties added in v0.11.0 

func (p Properties) UserProperties() iter.Seq2[string, string]

UserProperties yields every User Property in order.

func (Properties) UserProperty added in v0.11.0 

func (p Properties) UserProperty(key string) (string, bool)

UserProperty returns the first value of the User Property named key.

type ProtocolError added in v0.11.0 

type ProtocolError struct {
	// Reason is the DISCONNECT reason code sent: 0x81 Malformed Packet,
	// 0x82 Protocol Error, or a more specific code such as 0x93 Receive
	// Maximum exceeded.
	Reason wire.ReasonCode
	Packet wire.PacketType
	Detail string
}

ProtocolError is a broker packet that violates MQTT v5. The client sends DISCONNECT with Reason, closes the connection and reconnects.

func (*ProtocolError) Error added in v0.11.0 

func (e *ProtocolError) Error() string

type PublishMode 

type PublishMode uint8

PublishMode controls QoS 0 Client.Publish write-completion semantics. QoS 1/2 always wait for the broker's ack. 

const (
	// PublishFireAndForget (default): Publish returns once the
	// message is queued for the writer goroutine. Transport errors
	// are not surfaced.
	PublishFireAndForget PublishMode = iota

	// PublishWaitForFlush makes Publish block until the packet is
	// written to the connection and surfaces transport errors. On a TCP
	// or Unix connection with nothing queued for the writer goroutine the
	// caller writes the packet itself: the header and the caller's
	// payload go out in one writev, and the payload is not copied. ctx
	// bounds the call either way. A write it interrupts is finished by
	// the writer goroutine, which copies what is left, so the packet may
	// still reach the broker after Publish returned ctx's error; one
	// interrupted before its first byte is not sent. Useful as a
	// connection-health probe.
	PublishWaitForFlush
)

func (PublishMode) String 

func (m PublishMode) String() string

String returns a stable name for the mode.

type PublishOptions added in v0.11.0 

type PublishOptions struct {
	Topic   string
	Payload []byte
	QoS     byte // 0, 1 or 2
	Retain  bool

	// PayloadFormatIndicator 1 declares the payload UTF-8 text, 0 bytes.
	PayloadFormatIndicator *byte
	// MessageExpiryInterval is the message's lifetime in seconds; 0 is
	// valid (deliver now or never).
	MessageExpiryInterval *uint32
	ContentType           string
	ResponseTopic         string
	CorrelationData       []byte
	// TopicAlias sends an alias the broker already knows instead of the
	// topic (QoS 0 only); see also WithOutboundTopicAliases.
	TopicAlias     uint16
	UserProperties []UserProperty
}

PublishOptions is a message to publish. Topic is required unless a QoS 0 message uses TopicAlias; optional properties are sent only when set (a nil pointer or empty string or slice is absent).

func DecodePublish added in v0.11.0 

func DecodePublish(frame []byte) (PublishOptions, uint16, error)

DecodePublish is the inverse of EncodePublish: it returns the options of the PUBLISH packet in frame and its packet identifier. The options own their memory.

type PublisherQueue 

type PublisherQueue interface {
	// Enqueue stores e, assigning it the next Seq, which it returns.
	// When the queue already holds limit.Max entries it refuses with
	// ErrQueueFull or, under DropOldest, removes the oldest entries above
	// limit.Keep and returns them; the check, the removal and the insert
	// are one atomic step. It returns ErrQueueFull when there is nothing
	// it may remove.
	Enqueue(ctx context.Context, e QueueEntry, limit QueueLimit) (seq uint64, evicted []QueueEntry, err error)

	// Peek returns up to n entries with Seq above after, oldest first.
	Peek(ctx context.Context, after uint64, n int) ([]QueueEntry, error)

	// Ack removes the entry with seq. Acking a missing entry is not an
	// error.
	Ack(ctx context.Context, seq uint64) error

	// Len returns the number of entries, in constant time.
	Len(ctx context.Context) (int, error)

	// Close releases the queue's resources. Later calls return
	// ErrQueueClosed.
	Close() error
}

PublisherQueue stores the messages a QueuePublisher has accepted until the broker has them. Implementations must be safe for concurrent use and must not share memory with their callers: Enqueue copies the entry, and the entries they return belong to the caller. github.com/ashtonian/mqttv5/queue/queuetest checks an implementation against this contract.

type Queue 

type Queue[T any] struct {
	// contains filtered or unexported fields
}

Queue is an unbounded MPSC FIFO of T. Enqueue is non-blocking; Dequeue blocks until an item arrives, the queue is closed, or ctx cancels. After Close, queued items are still returned by Dequeue; once drained, Dequeue returns (zero, false).

func NewQueue 

func NewQueue[T any]() *Queue[T]

NewQueue returns a ready-to-use Queue.

Example

ExampleNewQueue shows the lock-protected unbounded MPSC queue returned by mqttv5.Client.SubscribeQueue. It is exported so callers can build their own internal pipelines on top of the same type. 

package main

import (
	"context"
	"fmt"

	"github.com/ashtonian/mqttv5"
)

func main() {
	q := mqttv5.NewQueue[string]()

	go func() {
		defer q.Close()
		q.Enqueue("one")
		q.Enqueue("two")
	}()

	ctx := context.Background()
	for {
		v, ok := q.Dequeue(ctx)
		if !ok {
			break
		}
		fmt.Println(v)
	}
}

Output:
one
two

func (*Queue[T]) Close 

func (q *Queue[T]) Close()

Close signals waiting Dequeue calls to drain. Items already queued remain available.

func (*Queue[T]) Dequeue 

func (q *Queue[T]) Dequeue(ctx context.Context) (T, bool)

Dequeue blocks until an item is available, the queue is closed, or ctx cancels. ok is false when no item could be retrieved.

func (*Queue[T]) Enqueue 

func (q *Queue[T]) Enqueue(item T) bool

Enqueue appends an item. Returns false if the queue is closed.

func (*Queue[T]) Len 

func (q *Queue[T]) Len() int

Len returns the current item count. Mostly for tests and metrics.

func (*Queue[T]) TryDequeue 

func (q *Queue[T]) TryDequeue() (T, bool)

TryDequeue removes and returns the head without blocking.

type QueueEntry 

type QueueEntry struct {
	// Seq is assigned by the queue at Enqueue. It increases strictly
	// within a queue and is never reused.
	Seq uint64

	// ID identifies the message for good. [QueuePublisher] stores it with
	// the session record of the message's exchange, so after a restart
	// it continues that exchange instead of starting another, and sends
	// it as the [QueueIDProperty] user property when
	// [WithQueueIdempotencyKey] is set.
	ID string

	// Publish is the message, QoS 1 or 2. PacketID and Dup are unused;
	// the Message Expiry Interval is carried by ExpiresAt.
	Publish PublishOptions

	EnqueuedAt time.Time

	// ExpiresAt is when the message may no longer be sent: the earlier of
	// its Message Expiry Interval and [WithQueueTTL], counted from
	// EnqueuedAt. Zero means never.
	ExpiresAt time.Time
}

QueueEntry is one message in a PublisherQueue.

type QueueLimit added in v0.11.0 

type QueueLimit struct {
	// Max is the most entries the queue may hold; 0 means no bound.
	Max int
	// Policy says what Enqueue does when the queue holds Max entries:
	// [DropNewest] refuses the new entry with [ErrQueueFull];
	// [DropOldest] removes the oldest entries with Seq above Keep to make
	// room.
	Policy DropPolicy
	// Keep is the highest Seq DropOldest must not remove: the entries up
	// to it are being published.
	Keep uint64
}

QueueLimit bounds a PublisherQueue for one Enqueue.

type QueueOption 

type QueueOption func(*queueConfig)

QueueOption customises a QueuePublisher.

func WithDeadLetter 

func WithDeadLetter(fn func(QueueEntry, error)) QueueOption

WithDeadLetter receives every message removed without the broker accepting it, with the reason: a refusal WithQueueClassifier calls permanent, ErrMessageExpired, or ErrQueueFull for an eviction. It is called from internal goroutines, possibly concurrently, before the message leaves the queue; it must not block.

func WithQueueClassifier added in v0.11.0 

func WithQueueClassifier(fn func(err error) bool) QueueOption

WithQueueClassifier decides which failed publishes are dead-lettered rather than retried: fn reports whether err is permanent. The default, PermanentPublishError, treats the broker's refusals for the message itself as permanent and everything else as passing.

func WithQueueDropPolicy 

func WithQueueDropPolicy(p DropPolicy) QueueOption

WithQueueDropPolicy decides what Publish does when the queue is at WithQueueMaxSize: DropNewest (default) returns ErrQueueFull; DropOldest removes the oldest messages not yet being published, passes them to the dead-letter callback with ErrQueueFull, and stores the new one.

func WithQueueIdempotencyKey added in v0.11.0 

func WithQueueIdempotencyKey() QueueOption

WithQueueIdempotencyKey adds each message's QueueEntry.ID as the QueueIDProperty user property, so consumers can recognise a message published again after a session loss.

func WithQueueMaxSize 

func WithQueueMaxSize(n int) QueueOption

WithQueueMaxSize bounds the queue at n entries; at the bound WithQueueDropPolicy decides. 0 (default) means unbounded.

func WithQueueRetryBackoff added in v0.11.0 

func WithQueueRetryBackoff(b Backoff) QueueOption

WithQueueRetryBackoff spaces the attempts of a message whose publish failed for a reason that may pass. Default DefaultQueueRetryBackoff.

func WithQueueTTL 

func WithQueueTTL(d time.Duration) QueueOption

WithQueueTTL limits how long a message may wait: it expires d after Publish (or earlier, at its own Message Expiry Interval). An expired message goes to the dead-letter callback with ErrMessageExpired; one that is sent carries the time it has left, in whole seconds rounded up, as its Message Expiry Interval.

func WithQueueWindow added in v0.11.0 

func WithQueueWindow(n int) QueueOption

WithQueueWindow sets how many messages are published at once. Default DefaultQueueWindow; the broker's Receive Maximum still applies on top.

type QueuePublisher 

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

QueuePublisher accepts QoS 1/2 messages into a PublisherQueue and publishes them in the background, many at a time.

QueuePublisher.Publish returns once the message is stored. A drain goroutine keeps up to WithQueueWindow messages in flight, in queue order, while the client is connected, and removes each from the queue when the broker has accepted it. A refusal that cannot pass (see WithQueueClassifier) and a message that expired go to the WithDeadLetter callback and are removed; other failures are retried with WithQueueRetryBackoff, after the messages behind them if need be, so one bad message never holds up the rest.

Each message has one MQTT exchange at a time. Its QueueEntry.ID is stored with the exchange in the client's session (WithStore), and the message is removed from the queue before the session record, so with a persistent queue and store a restarted process continues an interrupted exchange instead of publishing the message again: QoS 2 stays exactly-once across a crash, and QoS 1 is at most resent with DUP=1. When removing a finished message from the queue fails, the session keeps the exchange's record and the removal is retried with WithQueueRetryBackoff, in this process or after a restart, without publishing the message again. A message is published again as new only when the broker lost the session (see SessionLossPolicy).

The QueuePublisher publishes on the client's own connection, never through WithPublisherPool.

func NewQueuePublisher 

func NewQueuePublisher(client *Client, queue PublisherQueue, opts ...QueueOption) (*QueuePublisher, error)

NewQueuePublisher starts a QueuePublisher draining queue through client. Entries already in the queue are published in order once the client is connected; nothing is removed by construction.

Example

ExampleNewQueuePublisher queues publishes and drains them to the broker from a background goroutine, many at a time. Publish returns once the queue has stored the entry, before the broker round-trip.

To keep queued and in-flight messages across a process restart, swap mqttv5.NewMemoryPublisherQueue for the queue/file submodule and give the client a store/file session store with mqttv5.WithStore. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ashtonian/mqttv5"
)

func main() {
	cli, _ := mqttv5.New(mqttv5.WithBroker("mqtt://localhost:1883"))
	ctx := context.Background()
	_ = cli.Connect(ctx)
	defer cli.Disconnect(ctx)

	pub, err := mqttv5.NewQueuePublisher(cli, mqttv5.NewMemoryPublisherQueue(),
		mqttv5.WithQueueWindow(32),
		mqttv5.WithQueueTTL(24*time.Hour),
		mqttv5.WithDeadLetter(func(e mqttv5.QueueEntry, err error) {
			fmt.Println("dead letter", e.Publish.Topic, err)
		}),
	)
	if err != nil {
		panic(err)
	}
	defer pub.Close(context.Background())

	_ = pub.Publish(ctx, mqttv5.PublishOptions{
		Topic:   "events/durable",
		Payload: []byte("at-least-once"),
		QoS:     1,
	})
}

func (*QueuePublisher) Close 

func (p *QueuePublisher) Close(ctx context.Context) error

Close stops publishing, waits until ctx for the exchanges in flight to finish, and closes the queue. Messages whose exchange is still running stay in the queue and in the session; the next QueuePublisher on them continues those exchanges.

func (*QueuePublisher) Publish 

func (p *QueuePublisher) Publish(ctx context.Context, opts PublishOptions) error

Publish stores a copy of opts in the queue and returns; the drain publishes it. The message must be valid to send (QoS 1 or 2, a topic name, no Topic Alias): an invalid one is refused here rather than dead-lettered later.

func (*QueuePublisher) PublishValue added in v0.12.0 

func (p *QueuePublisher) PublishValue(ctx context.Context, opts PublishOptions, v any) error

PublishValue encodes v with the client's Encoding and queues it as QueuePublisher.Publish does. An encoding failure is returned here, never dead-lettered.

type ReasonCode added in v0.11.0 

type ReasonCode = wire.ReasonCode

ReasonCode is an MQTT v5 reason code (§2.4). Codes of 0x80 and above are errors; see IsError and String.

type ReasonCodeError added in v0.11.0 

type ReasonCodeError struct {
	Packet          wire.PacketType
	Code            wire.ReasonCode
	ReasonString    string
	UserProperties  []wire.UserProperty
	ServerReference string
}

ReasonCodeError is a broker's refusal: a reason code of 0x80 or above on CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK or DISCONNECT, with the optional diagnostics the broker attached.

errors.Is matches it against ErrConnectRefused (for CONNACK) and against ErrNotAuthorized, ErrQuotaExceeded, ErrTopicNameInvalid, ErrPayloadFormatInvalid and ErrPacketTooLarge by code. Use errors.As to read the code itself.

func (*ReasonCodeError) Error added in v0.11.0 

func (e *ReasonCodeError) Error() string

func (*ReasonCodeError) Is added in v0.11.0 

func (e *ReasonCodeError) Is(target error) bool

Is reports whether e corresponds to target.

type RejectedByDisconnectError added in v0.11.0 

type RejectedByDisconnectError struct {
	Packet     PacketType // SUBSCRIBE or UNSUBSCRIBE
	Attempts   int
	Disconnect DisconnectInfo // the last of them
}

RejectedByDisconnectError reports a SUBSCRIBE or UNSUBSCRIBE the client gave up on: each of the Attempts times it was sent, the broker ended the connection with a DISCONNECT blaming a packet it received instead of answering. Some brokers do this for limits they do not announce, such as mosquitto's 200 topic levels. A Subscribe that fails this way has closed its subscription.

func (*RejectedByDisconnectError) Error added in v0.11.0 

func (e *RejectedByDisconnectError) Error() string

type ServerFinalVerifier added in v0.10.0 

type ServerFinalVerifier interface {
	VerifyServerFinal(serverData []byte) error
}

ServerFinalVerifier is an optional interface an Authenticator may also implement to verify the server's concluding AuthenticationData — the CONNACK on the initial handshake, or the AUTH 0x00 Success that ends a re-authentication (§4.12). For mechanisms with mutual authentication (e.g. SCRAM, where the server proves knowledge of the credential via a server signature) this closes the loop; a non-nil error aborts the connect, or fails Reauthenticate and tears the connection down.

The serverData is whatever AuthenticationData the broker attached to the terminal packet (possibly empty). Authenticators whose mechanism has no server proof (e.g. a bearer token) simply do not implement this interface, and the terminal data is ignored. VerifyServerFinal is called on the same goroutine as the corresponding Continue.

type ServerRedirect added in v0.11.0 

type ServerRedirect struct {
	Packet    PacketType // CONNACK or DISCONNECT
	Reason    ReasonCode
	Reference string // as the broker sent it
}

ServerRedirect is a broker's request that the client use another server (§4.11): a CONNACK or DISCONNECT with reason 0x9C Use another server or 0x9D Server moved, carrying a Server Reference.

func (ServerRedirect) Permanent added in v0.11.0 

func (r ServerRedirect) Permanent() bool

Permanent reports a move (0x9D Server moved) rather than a temporary redirection (0x9C Use another server).

type SessionLossPolicy added in v0.11.0 

type SessionLossPolicy uint8

SessionLossPolicy decides what happens to unacknowledged QoS 1/2 publishes when a connection starts without the old session: the broker answered CONNACK with Session Present = 0 (restart without persistence, session expiry, failover to another broker in WithBrokers) or the client connected with CleanStart=1. 

const (
	// SessionLossRepublish sends them again as new messages (DUP=0) in
	// their original order; their Publish calls keep waiting for the new
	// acknowledgement. Delivery stays at-least-once, and a message the
	// old broker had already forwarded is delivered twice — including
	// QoS 2, whose exactly-once guarantee cannot survive a session loss.
	// Messages whose Message Expiry Interval ran out complete with
	// [ErrMessageExpired]; the rest are sent with the remaining interval.
	SessionLossRepublish SessionLossPolicy = SessionLossPolicy(inflight.Republish)

	// SessionLossFail completes them with [ErrSessionLost] and leaves
	// any retry to the caller.
	SessionLossFail SessionLossPolicy = SessionLossPolicy(inflight.Fail)
)

type Stats 

type Stats struct {
	Connects          uint64 // successful CONNECT/CONNACK handshakes
	ConnectFailures   uint64 // CONNECT attempts that failed (any cause)
	Disconnects       uint64 // connections lost, for any reason but Disconnect
	ServerDisconnects uint64 // broker-sent DISCONNECT packets
	PingTimeouts      uint64 // PINGRESP timeouts that forced a reconnect

	PublishesSent     uint64 // Publish calls that returned nil (QoS 1/2 once acknowledged), and QueuePublisher messages the broker accepted
	PublishesAcked    uint64 // PUBACK and PUBCOMP packets received, refusals included
	PublishesInflight uint64 // QoS 1/2 outbound entries awaiting ack
	SendQuota         uint64 // QoS 1/2 PUBLISHes the broker's Receive Maximum still allows in flight
	PublishesReplayed uint64 // QoS 1/2 replays issued after reconnect

	InboundPublishes uint64 // PUBLISHes received from the broker
	InboundDropped   uint64 // inbound PUBLISHes dropped due to full subscriber buffer
	InboundUnrouted  uint64 // inbound PUBLISHes no subscription matched (acked and dropped)
	DecodeFailures   uint64 // typed-subscription payloads the Encoding rejected (acked, not delivered)

	SubscribesSent      uint64 // SUBSCRIBE packets emitted
	UnsubscribesSent    uint64 // UNSUBSCRIBE packets emitted
	SubscriptionsActive uint64 // count of currently registered subscriptions

	PoolFallbacks uint64 // publishes no pool member could send, handed to the main connection
	PoolConnected uint64 // publisher pool members with a live connection

	AcksIgnored    uint64 // PUBACK/PUBREC/PUBCOMP for an unknown or mismatched packet identifier
	StoreErrors    uint64 // failed session Store operations (see WithStore)
	ProtocolErrors uint64 // broker protocol violations that closed the connection
}

Stats is the snapshot of Client counters returned by Client.Stats.

Two flavours of field:

  • Monotonic counters (Connects, ConnectFailures, Disconnects, ServerDisconnects, PingTimeouts, PublishesSent, PublishesAcked, PublishesReplayed, InboundPublishes, InboundDropped, InboundUnrouted, DecodeFailures, SubscribesSent, UnsubscribesSent, PoolFallbacks, AcksIgnored, StoreErrors, ProtocolErrors). They start at zero in New and are never reset: they span every Connect and Disconnect of the Client. Sampled atomically — no locking on the read side.
  • Point-in-time gauges (PublishesInflight, SendQuota, SubscriptionsActive, PoolConnected) read live state under a mutex (the session state and the subscriptions map respectively). The locks are cheap but a scraper hitting Stats() at very high frequency does contend with Subscribe / Unsubscribe and the ack path. Default to sampling at most a few times per second.

With WithPublisherPool every field covers the pool members' connections as well as the main one: publishes made through the pool count as the Client's, and a pool member's reconnects count with the main connection's.

The zero value is returned when WithStats was not set on the Config — when stats are disabled the hot path skips every atomic increment, so leave WithStats off when you have no scraper.

type StoreError added in v0.11.0 

type StoreError = inflight.StoreError

StoreError reports a failed write to the WithStore store: Op names the write and Err is the store's error. errors.Is matches both ErrStoreFailed and Err.

type SubscribeError added in v0.11.0 

type SubscribeError struct {
	Results []SubscribeResult
}

SubscribeError reports filters the broker refused. Results holds the outcome of every filter of the SUBSCRIBE. When some filters were granted the subscription is active for those; when none were, its channel or queue is closed.

func (*SubscribeError) Error added in v0.11.0 

func (e *SubscribeError) Error() string

func (*SubscribeError) Is added in v0.11.0 

func (e *SubscribeError) Is(target error) bool

Is matches the sentinel of any refusal's reason code (see ReasonCodeError).

type SubscribeOption 

type SubscribeOption func(*subscribeConfig)

SubscribeOption configures client-side delivery for one Subscribe call. Protocol flags (QoS, NoLocal, etc.) live on TopicFilter itself — these options control what happens after delivery.

func SubAutoAck 

func SubAutoAck() SubscribeOption

SubAutoAck makes Client.Subscribe / Client.SubscribeQueue ack each delivery on the read goroutine before handing it to the consumer; Message.Ack is then a no-op.

Auto-ack gives up at-least-once processing: a consumer that crashes between delivery and processing has nothing to replay, because the broker already considers the message delivered. It also gives up the broker's flow control, so a full buffer or queue drops QoS 1 and 2 messages as it drops QoS 0. Use it for QoS 0 or observational consumers; keep manual ack for ledger-style workloads.

Ignored by Client.SubscribeCallback (already auto-acks).

func SubBuffer 

func SubBuffer(n int) SubscribeOption

SubBuffer sets Client.Subscribe's channel buffer size: how many QoS 0 messages the channel holds before it drops one. QoS 1 and 2 messages have room beyond it (see Client.Subscribe). Default DefaultSubscribeBuffer. Ignored by Client.SubscribeQueue and Client.SubscribeCallback.

func SubDropPolicy 

func SubDropPolicy(p DropPolicy) SubscribeOption

SubDropPolicy chooses behaviour when the channel buffer or queue cap fills.

  • Subscribe (chan): only DropNewest is supported. Passing DropOldest here causes Subscribe to return ErrChanDropOldestUnsupported — peek-and-pop on a consumer-owned channel would race the receiver. Use SubscribeQueue for DropOldest semantics.
  • SubscribeQueue: both policies are honored. DropOldest keeps the newest messages up to the cap whatever their QoS, so it may drop QoS 1 and 2 messages too.
  • SubscribeCallback: ignored (callback never drops; slow handler stalls the connection instead).

Default is the client-level WithDropPolicy (DropNewest).

func SubEncoding added in v0.12.0 

func SubEncoding(e Encoding) SubscribeOption

SubEncoding decodes one typed subscription with e instead of the client's Encoding. Ignored by untyped subscriptions.

func SubMaxQueueSize 

func SubMaxQueueSize(n int) SubscribeOption

SubMaxQueueSize caps Client.SubscribeQueue's length for this subscription. QoS 1 and 2 messages have room beyond the cap (see Client.SubscribeQueue). UnboundedQueue removes the cap; 0 keeps the client default from WithMaxSubscribeQueueSize (DefaultMaxSubscribeQueueSize).

func SubOnDecodeError added in v0.12.0 

func SubOnDecodeError(fn func(*Message, error)) SubscribeOption

SubOnDecodeError is called with each message of a typed subscription whose payload the Encoding rejects, and the error. Such a message is not delivered: it is acked, so the broker does not send it again, and counted in Stats.DecodeFailures. fn runs on the read goroutine before the ack, so it must not block, nor call Client.Disconnect.

func SubOnDrop 

func SubOnDrop(fn func(*Message)) SubscribeOption

SubOnDrop fires when a message is dropped by Client.Subscribe (buffer full) or Client.SubscribeQueue (cap reached). It runs on the read goroutine before the dropped message is acknowledged, so every field is readable; it must not block, nor call Client.Disconnect. On a SubZeroCopy subscription the fields must not be retained past return.

func SubZeroCopy added in v0.11.0 

func SubZeroCopy() SubscribeOption

SubZeroCopy delivers Messages whose Topic, Payload and Properties alias the pooled network frame instead of an owned copy, saving one copy of the packet per message. The fields are valid only until the Message is acked by every subscription it reached (for Client.SubscribeCallback, until the callback returns); use Message.CloneTopic / Message.ClonePayload to keep them longer. QoS 0 messages must also be acked so the frame returns to the pool. A PUBLISH that also matches a subscription without SubZeroCopy is delivered as an owned copy to all of them. Ignored together with SubAutoAck, which acks before delivery and therefore needs an owned copy.

type SubscribeResult added in v0.11.0 

type SubscribeResult struct {
	Filter TopicFilter
	// Reason is the SUBACK reason code: 0x00–0x02 grant that QoS,
	// 0x80 and above refuse the filter.
	Reason wire.ReasonCode
}

SubscribeResult is the broker's answer for one filter of a SUBSCRIBE.

func (SubscribeResult) Granted added in v0.11.0 

func (r SubscribeResult) Granted() bool

Granted reports whether the broker accepted the filter.

func (SubscribeResult) QoS added in v0.11.0 

func (r SubscribeResult) QoS() byte

QoS is the maximum QoS the broker granted; meaningful when Granted.

type SubscriptionToken 

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

SubscriptionToken identifies a subscription for Client.Unsubscribe and reports what the broker decided.

func (SubscriptionToken) Err added in v0.11.0 

func (t SubscriptionToken) Err() error

Err reports why the subscription ended other than by Unsubscribe or Disconnect: the broker refused every filter, initially or when the client re-subscribed after a session loss. Nil while it is active.

func (SubscriptionToken) Results added in v0.11.0 

func (t SubscriptionToken) Results() []SubscribeResult

Results returns the broker's answer per filter from the subscription's SUBACK, or nil before it arrived.

type TopicFilter 

type TopicFilter struct {
	Topic             string
	QoS               byte // maximum QoS: 0, 1 or 2
	NoLocal           bool
	RetainAsPublished bool
	RetainHandling    byte // 0 send retained, 1 only for a new subscription, 2 never
}

TopicFilter is one entry of a SUBSCRIBE: a topic filter and its subscription options (§3.8.3.1). Only Topic is required. 

{Topic: "events/#"}                        // QoS 0
{Topic: "events/#", QoS: 1}                // at-least-once
{Topic: "events/#", QoS: 1, NoLocal: true} // don't echo own publishes back

type TypedMessage 

type TypedMessage[T any] struct {
	*Message
	Topic string
	Value T
}

TypedMessage is a delivered message with its payload decoded. Value and Topic are safe to retain past Ack, including on SubZeroCopy subscriptions (where Topic is cloned off the frame); the embedded *Message follows the usual ownership rules for Payload and Properties.

type UnsubscribeError added in v0.11.0 

type UnsubscribeError struct {
	Results []UnsubscribeResult
}

UnsubscribeError reports topic filters the broker refused to unsubscribe. The local subscription is closed regardless; the broker may keep delivering messages for them, which the client then drops.

func (*UnsubscribeError) Error added in v0.11.0 

func (e *UnsubscribeError) Error() string

type UnsubscribeResult added in v0.11.0 

type UnsubscribeResult struct {
	Topic string
	// Reason is the UNSUBACK reason code: 0x00 success, 0x11 there was no
	// such subscription (not an error), 0x80 and above refused.
	Reason wire.ReasonCode
}

UnsubscribeResult is the broker's answer for one topic filter of an UNSUBSCRIBE.

type UserProperty added in v0.11.0 

type UserProperty = wire.UserProperty

UserProperty is one User Property name/value pair. A packet may carry the same name more than once.

type WillOptions added in v0.11.0 

type WillOptions struct {
	Topic   string
	Payload []byte
	QoS     byte
	Retain  bool

	// DelayInterval delays the Will by this many seconds after the
	// connection ends; nil or 0 publishes it at once.
	DelayInterval          *uint32
	PayloadFormatIndicator *byte
	MessageExpiryInterval  *uint32
	ContentType            string
	ResponseTopic          string
	CorrelationData        []byte
	UserProperties         []UserProperty
}

WillOptions is the Will Message the broker publishes for the client when its connection ends without a DISCONNECT (§3.1.3.2).

type WriteOverflowPolicy 

type WriteOverflowPolicy uint8

WriteOverflowPolicy controls what QoS 0 Client.Publish does when the internal writer queue (sized by WithWriteQueueSize) is full. QoS 1/2 are unaffected — they always block on the broker ack. 

const (
	// WriteBlock (default) blocks Publish until the writer queue has
	// room, the call's ctx fires, or the connection goes down. The
	// caller surfaces overload as latency.
	WriteBlock WriteOverflowPolicy = iota

	// WriteDropNewest returns [ErrWriteQueueFull] immediately when
	// the queue has no room. Use for high-rate fire-and-forget
	// telemetry where the producer goroutine must not block.
	WriteDropNewest
)

func (WriteOverflowPolicy) String 

func (p WriteOverflowPolicy) String() string

String returns a stable name for the policy.

Source Files

View all Source files

Directories

Path Synopsis
codec
json module
msgpack module
internal
clock
Package clock abstracts time so timing-dependent code (keep-alive, backoff, queue draining) can be tested without sleeping.
 Package clock abstracts time so timing-dependent code (keep-alive, backoff, queue draining) can be tested without sleeping.
inflight
Package inflight is the client's MQTT v5 session state machine: packet identifier ownership, outbound QoS 1/2 flows, inbound QoS 1/2 flows and the acknowledgements owed to the broker.
 Package inflight is the client's MQTT v5 session state machine: packet identifier ownership, outbound QoS 1/2 flows, inbound QoS 1/2 flows and the acknowledgements owed to the broker.
soak
Package soak sizes the randomized tests.
 Package soak sizes the randomized tests.
testbroker
Package testbroker is a scripted MQTT v5 peer for tests.
 Package testbroker is a scripted MQTT v5 peer for tests.
trie
Package trie implements an MQTT v5 topic filter matcher used by the client to dispatch inbound PUBLISH packets to subscribed handlers.
 Package trie implements an MQTT v5 topic filter matcher used by the client to dispatch inbound PUBLISH packets to subscribed handlers.
filedb module
queue
queuetest
Package queuetest checks a mqttv5.PublisherQueue implementation against the interface's contract.
 Package queuetest checks a mqttv5.PublisherQueue implementation against the interface's contract.
file module
Package session defines how an MQTT v5 client persists its session state: the Store contract, the [Record]s it holds, and MemoryStore, an in-memory reference implementation.
 Package session defines how an MQTT v5 client persists its session state: the Store contract, the [Record]s it holds, and MemoryStore, an in-memory reference implementation.
storetest
Package storetest checks that a session.Store implementation honours the contract the client relies on.
 Package storetest checks that a session.Store implementation honours the contract the client relies on.
store
file module
Package transport abstracts the network layer underneath the MQTT v5 client.
 Package transport abstracts the network layer underneath the MQTT v5 client.
ws module
Package wire implements the MQTT v5 wire-format codec.
 Package wire implements the MQTT v5 wire-format codec.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.