Help us shape the future of Go! Take the 2026 Go Developer Survey

mqttclient

package module
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: BSD-3-Clause Imports: 19 Imported by: 0

README

mqtt-client

An MQTT 3.1.1 / 5.0 client for Go, modelled on the libmosquitto client API and built on the comqtt packet codec.

c, err := mqttclient.New(mqttclient.Options{
	// Tried in order until one accepts, cycling; each server has its own
	// connection settings.
	Servers: []mqttclient.Server{{
		Address:         "localhost:1883",
		ProtocolVersion: mqttclient.MQTT5,
		ClientID:        "example",
		CleanStart:      true,
		KeepAlive:       60,
	}},
}, mqttclient.Handlers{
	OnDisconnect: func(_ *mqttclient.Client, ev mqttclient.DisconnectEvent) {
		log.Println("disconnected:", ev.Err)
	},
	OnMessage: func(_ *mqttclient.Client, m *mqttclient.Message) {
		log.Printf("%s: %s", m.Topic, m.Payload)
	},
	OnConnect: func(c *mqttclient.Client, ack mqttclient.ConnAck) {
		// Subscriptions are not resent on a new connection: subscribe here.
		_, _ = c.Subscribe(context.Background(), []mqttclient.Subscription{{Topic: "a/#", QoS: 1}}, nil)
	},
})
if err != nil {
	log.Fatal(err)
}
// Either keep the connection up, reconnecting as needed (blocks until ctx
// ends or Disconnect):
//     go func() { log.Println("run:", c.Run(ctx)) }()
// or connect once:
if _, err := c.Connect(ctx); err != nil {
	log.Fatal(err)
}
defer c.Disconnect(ctx, 0, nil)

p, err := c.Publish(ctx, &mqttclient.Message{Topic: "a/b", Payload: []byte("hi"), QoS: 1})
if err != nil {
	log.Fatal(err)
}
if _, err := p.Wait(ctx); err != nil { // PUBACK received, or refused
	log.Fatal(err)
}

Testing

go test -race ./...
(cd integration && go test -race ./...)   # starts its own mosquitto; skips without one
./conformance/run.sh      # libmosquitto's client test suite; needs python3 and
                          # a mosquitto checkout (MOSQUITTO_SRC, default ../mosquitto)

For manual tests against a real broker, see cmd/mqttconnect (connect and disconnect), cmd/mqttpub (publish), cmd/mqttsub (subscribe and unsubscribe) and cmd/interactiveclient (a long-running app on Run that reconnects by itself).

To probe how a broker handles packets a client should never send, see cmd/mqttfuzz. Its engine is the fuzz package, so a test can run the same YAML scenarios in-process:

sc, err := fuzz.LoadFS(fuzz.Scenarios, "scenarios/double-connect.yaml", nil)
if err != nil {
	t.Fatal(err)
}
res, err := sc.Run(ctx, fuzz.RunOptions{Address: addr, Username: "u", Password: "p", Logf: t.Logf})
switch {
case errors.Is(err, fuzz.ErrSetup): // the run could not happen (broker unreachable, ...)
	t.Fatal(err)
case errors.Is(err, fuzz.ErrExpectation): // the broker did not behave as the scenario asserts
	t.Errorf("%v\n%s", err, res.JSON())
}

Using it

go get github.com/AUTOSOLN/mqtt-client

Go applies replace directives only in the main module, so add this to your own go.mod to build against the AUTOSOLN comqtt fork:

replace github.com/wind-c/comqtt/v2 => github.com/AUTOSOLN/comqtt/v2 <pseudo-version>

Without the replace, the client still builds and works against upstream github.com/wind-c/comqtt/v2.

License

BSD-3-Clause. Portions are derived from Eclipse Mosquitto (used under EDL-1.0 / BSD-3-Clause) and comqtt / mochi-mqtt (MIT); derived files carry attribution headers.

Documentation

Overview

Package mqttclient is an MQTT 3.1.1 / 5.0 client modelled on the libmosquitto client API, using the comqtt packets codec for wire encoding and decoding.

See PLAN.md for the design and roadmap.

Index

Constants

View Source 
const (
	MQTT31  byte = 3
	MQTT311 byte = 4
	MQTT5   byte = 5
)

Protocol versions accepted in Options.ProtocolVersion.

View Source 
const (
	DefaultReceiveMaximum uint16 = 20
	DefaultMaxInflight    uint16 = 20
)

Defaults match libmosquitto (mosquitto_reinitialise).

Variables

View Source 
var (
	ErrInvalid          = errors.New("mqttclient: invalid argument") // MOSQ_ERR_INVAL
	ErrNotSupported     = errors.New("mqttclient: not supported")    // MOSQ_ERR_NOT_SUPPORTED
	ErrNoConn           = errors.New("mqttclient: not connected")    // MOSQ_ERR_NO_CONN
	ErrAlreadyConnected = errors.New("mqttclient: already connected or connecting")
	ErrConnectionLost   = errors.New("mqttclient: connection lost")                    // MOSQ_ERR_CONN_LOST / MOSQ_ERR_ERRNO
	ErrKeepalive        = errors.New("mqttclient: keepalive timeout")                  // MOSQ_ERR_KEEPALIVE
	ErrProtocol         = errors.New("mqttclient: protocol error")                     // MOSQ_ERR_PROTOCOL
	ErrMalformedPacket  = errors.New("mqttclient: malformed packet")                   // MOSQ_ERR_MALFORMED_PACKET
	ErrOversizePacket   = errors.New("mqttclient: packet exceeds maximum packet size") // MOSQ_ERR_OVERSIZE_PACKET
	ErrServerDisconnect = errors.New("mqttclient: server sent DISCONNECT")
	ErrDisconnected     = errors.New("mqttclient: disconnected by the application")
	ErrQoSNotSupported  = errors.New("mqttclient: qos not supported by the server") // MOSQ_ERR_QOS_NOT_SUPPORTED
	ErrNoMid            = errors.New("mqttclient: all message ids are in use")
	ErrRunning          = errors.New("mqttclient: Run is managing the connection")
)

Sentinel errors, roughly following libmosquitto's MOSQ_ERR_* values. Errors returned by this package wrap one of these, so test with errors.Is.

Functions

func ReasonCodeString 

func ReasonCodeString(code byte) string

ReasonCodeString returns the MQTT 5 specification text for a reason code. Codes 0x00-0x02 have several meanings depending on the packet type; the most general one is returned.

Types

type Client 

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

Client is an MQTT client connection to one of the servers in Options.Servers at a time. It is safe for concurrent use.

func New 

func New(opts Options, h Handlers) (*Client, error)

New validates opts and returns a disconnected client (mosquitto_new).

func (*Client) ClientID 

func (c *Client) ClientID() string

ClientID returns the client identifier for the current server, which may have been assigned by an MQTT 5 server in CONNACK.

func (*Client) Connect 

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

Connect opens a network connection, sends CONNECT and waits for CONNACK or for ctx to end. OnConnect is called when CONNACK arrives; if the connection is refused, OnDisconnect follows.

Connect tries each server once, in order, starting from the current one (see ServerIndex), and stops at the first that accepts the connection. Each attempt is also bounded by that server's ConnectTimeout. If every attempt fails, the server after the last one tried becomes current, and Connect returns the error from a single server as it is (a *ConnRefusedError for a refusal), or the errors from several servers joined.

Connect does not reconnect after the connection is lost; OnDisconnect reports the loss and Connect may be called again. Use Run for a connection that is kept up automatically.

func (*Client) Disconnect 

func (c *Client) Disconnect(ctx context.Context, reason byte, props *Properties) error

Disconnect sends DISCONNECT with the given MQTT 5 reason code and properties, closes the connection and waits for it to shut down (mosquitto_disconnect_v5). For MQTT 3.x, reason must be 0 and props nil. OnDisconnect is called with a nil Err.

With Run, Disconnect also stops Run, which returns nil; it does not wait for Run to return. Between connections, Disconnect only stops Run.

func (*Client) IsConnected 

func (c *Client) IsConnected() bool

IsConnected reports whether the client has an accepted connection.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, m *Message) (*Pending, error)

Publish sends m (mosquitto_publish_v5). QoS 0 requires a connection and is complete once written. QoS 1 and 2 messages are queued: they are sent when the send quota allows, and are kept across connections until acknowledged, so a message published while disconnected is sent after the next successful Connect. ctx bounds the network write, not the acknowledgement; use the returned Pending to wait for that.

As in libmosquitto, the retain flag is cleared when the server does not support retain, and a QoS above the server's Maximum QoS fails with ErrQoSNotSupported. A message larger than the server's Maximum Packet Size fails with ErrOversizePacket.

func (*Client) Run 

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

Run connects and keeps the client connected: when the connection is lost or an attempt fails, it waits (see Server.ReconnectDelay) and connects again, until ctx ends or Disconnect is called. It blocks, so start it on its own goroutine.

Run starts with the current server (see ServerIndex). After a failed attempt it moves on to the next server in Options.Servers, cycling back to the first after the last. After an accepted connection is lost it tries the same server again first. The wait before each attempt uses the settings of the server about to be tried. A server that fails in a way retrying cannot fix (see below) is not tried again during this Run.

Each new connection resends unacknowledged QoS 1 and 2 messages, as Connect does, whichever server it is to. Subscriptions are not resent: subscribe from OnConnect, which runs for every accepted connection. OnConnectError reports each failed attempt.

Run returns:

  • nil after Disconnect;
  • ctx.Err() when ctx ends, after a clean DISCONNECT if connected (so the server does not publish the Will);
  • once every server has failed in a way retrying cannot fix, the error from a single server as it is, or the errors from several servers joined: a *ConnRefusedError for credentials, authorization, client id, protocol version and other non-transient reasons, a protocol error from the server (ErrProtocol), or a TLS certificate verification failure.

Connect returns ErrRunning while Run is active.

func (*Client) ServerIndex 

func (c *Client) ServerIndex() int

ServerIndex returns the index in Options.Servers of the current server: the one the client is connected to, or is connecting to, or will try next. It starts at 0. OnPreConnect may call it to learn which server is about to be tried.

func (*Client) SetCredentials 

func (c *Client) SetCredentials(username string, password []byte) error

SetCredentials replaces the username and password sent in the next CONNECT to the current server (mosquitto_username_pw_set). Call it from OnPreConnect to change them for the server about to be tried.

func (*Client) SetWill 

func (c *Client) SetWill(w *Message) error

SetWill replaces the Will sent in the next CONNECT to the current server; nil clears it (mosquitto_will_set_v5 / mosquitto_will_clear). Call it from OnPreConnect to change it for the server about to be tried.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, subs []Subscription, props *Properties) (*Pending, error)

Subscribe sends SUBSCRIBE for subs (mosquitto_subscribe_multiple, with options per filter). It needs a connection: subscriptions are not queued or resent, so subscribe again from OnConnect after reconnecting. The Pending completes on SUBACK with one granted QoS or failure reason code per filter. props may carry a Subscription Identifier and User Properties (MQTT 5).

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ctx context.Context, topics []string, props *Properties) (*Pending, error)

Unsubscribe sends UNSUBSCRIBE for topics (mosquitto_unsubscribe_multiple). It needs a connection. The Pending completes on UNSUBACK with one reason code per filter; MQTT 3.x has none, so they are all reported as 0.

type ConnAck 

type ConnAck struct {
	ReasonCode     byte // 0 on success; a 3.x return code or a 5 reason code otherwise
	SessionPresent bool
	Properties     *Properties // MQTT 5 only
}

ConnAck describes a received CONNACK.

type ConnRefusedError 

type ConnRefusedError struct {
	ReasonCode      byte
	ProtocolVersion byte
}

ConnRefusedError is returned by Connect, and reported to OnDisconnect, when the server answers CONNECT with a non-zero CONNACK return/reason code.

func (*ConnRefusedError) Error 

func (e *ConnRefusedError) Error() string

type DisconnectEvent 

type DisconnectEvent struct {
	// Err is nil when the application called Disconnect. Otherwise it wraps
	// one of ErrConnectionLost, ErrKeepalive, ErrProtocol,
	// ErrMalformedPacket, ErrServerDisconnect, or is a *ConnRefusedError.
	Err error

	// ReasonCode and Properties are set when the server sent DISCONNECT
	// (MQTT 5).
	ReasonCode byte
	Properties *Properties
}

DisconnectEvent describes why a connection ended.

type Handlers 

type Handlers struct {
	// OnPreConnect is called just before each network connection attempt,
	// synchronously from Connect or Run (mosquitto_pre_connect_callback_set).
	// ServerIndex reports the server about to be tried, and SetCredentials
	// and SetWill change what is sent to it in CONNECT.
	OnPreConnect func(c *Client)

	// OnConnect is called when CONNACK is received, whether the connection
	// was accepted or refused.
	OnConnect func(c *Client, ack ConnAck)

	// OnDisconnect is called once for every network connection that was
	// established, after it has closed.
	OnDisconnect func(c *Client, ev DisconnectEvent)

	// OnConnectError is called by Run for every connection attempt that
	// fails, whichever server it was to: the network connection, the TLS
	// handshake, CONNACK not arriving within ConnectTimeout, or a refusal
	// (which OnConnect and OnDisconnect also report).
	OnConnectError func(c *Client, err error)

	// OnMessage is called for every message received from the server: QoS 0
	// and 1 on arrival (after PUBACK is sent), QoS 2 when PUBREL arrives
	// (after PUBCOMP is sent). The handler owns m.
	OnMessage func(c *Client, m *Message)

	// OnPublish is called when a published message is complete: QoS 0 once
	// written to the network, QoS 1 on PUBACK, QoS 2 on PUBCOMP, or on a
	// PUBREC with a failure reason code (MQTT 5). reason and props are the
	// acknowledgement's reason code and properties.
	OnPublish func(c *Client, mid uint16, reason byte, props *Properties)

	// OnSubscribe is called when SUBACK arrives, with one granted QoS or
	// reason code per topic filter, in request order.
	OnSubscribe func(c *Client, mid uint16, granted []byte, props *Properties)

	// OnUnsubscribe is called when UNSUBACK arrives, with one reason code per
	// topic filter. MQTT 3.x has no reason codes, so they are all 0 (success).
	OnUnsubscribe func(c *Client, mid uint16, reasons []byte, props *Properties)
}

Handlers are the client callbacks. They are called one at a time, in order, on a goroutine owned by the client, the way libmosquitto calls callbacks from its network loop. Callbacks may call Client methods, but must not block waiting for network activity (for example waiting for an acknowledgement), since later callbacks queue behind them.

type Message 

type Message struct {
	Topic      string
	Payload    []byte
	QoS        byte
	Retain     bool
	Properties *Properties // MQTT 5 only

	// Mid is the packet identifier of a received QoS 1 or 2 message. It is
	// ignored by Publish and SetWill.
	Mid uint16

	// Dup is the DUP flag of a received message: the server marked it as a
	// possible redelivery. For QoS 2 it is the flag of the last PUBLISH
	// received before PUBREL. It is ignored by Publish and SetWill, which
	// set DUP themselves on a resend.
	Dup bool
}

Message is an application message: a PUBLISH payload, or a Will.

type Options 

type Options struct {
	// Servers are the brokers the client may connect to, each with its own
	// connection settings. Connect and Run try them in order, starting from
	// the current server (see Client.ServerIndex), and move on to the next
	// one, cycling back to the first, when an attempt fails. The session
	// (queued and in-flight QoS 1 and 2 messages) belongs to the client, so
	// it carries over from one server to the next.
	Servers []Server

	// Logger receives debug output. Nil discards it.
	Logger *slog.Logger

	// NoTopicCheck turns off the topic name and topic filter checks (empty
	// topic, UTF-8 rules, wildcard placement) for PUBLISH, SUBSCRIBE,
	// UNSUBSCRIBE, the Will and received PUBLISH, so that a malformed topic
	// goes to the server as given. Only the 65535-byte limit remains. It is
	// meant for testing how a server handles invalid topics.
	NoTopicCheck bool
}

Options configures a Client. At least one server is required.

type Pending 

type Pending struct {
	Mid uint16
	// contains filtered or unexported fields
}

Pending tracks a request (Publish, Subscribe or Unsubscribe) until it is complete.

func (*Pending) Done 

func (p *Pending) Done() <-chan struct{}

Done is closed when the message is complete.

func (*Pending) Wait 

func (p *Pending) Wait(ctx context.Context) (Result, error)

Wait blocks until the request is complete or ctx ends. It returns a *ReasonCodeError, with the Result, if the server refused the message or a topic filter (reason code 0x80 or above; for several filters, the first refused one). A ctx error does not cancel the request: a published message stays queued and is still sent and resent on later connections. A Subscribe or Unsubscribe fails with the connection's error if the connection closes before the acknowledgement.

type Properties 

type Properties = packets.Properties

Properties holds MQTT 5 properties. Several values only take effect when their matching flag is also set (SessionExpiryIntervalFlag, RequestProblemInfoFlag, PayloadFormatFlag, ...).

type ReasonCodeError 

type ReasonCodeError struct {
	Packet     string // the acknowledgement, for example "PUBACK"
	ReasonCode byte
}

ReasonCodeError is returned by Pending.Wait when the server answers with an MQTT 5 failure reason code (0x80 or above).

func (*ReasonCodeError) Error 

func (e *ReasonCodeError) Error() string

type Result 

type Result struct {
	// ReasonCode is the reason code of a publish's final acknowledgement
	// (MQTT 5); 0 for QoS 0.
	ReasonCode byte

	// ReasonCodes has one entry per topic filter of a Subscribe (the granted
	// QoS, or a failure code of 0x80 or above) or an Unsubscribe.
	ReasonCodes []byte

	Properties *Properties // properties of the acknowledgement (MQTT 5)
}

Result is the server's answer to a request.

type Server 

type Server struct {
	// Address is the broker as host:port, where host is an IP address or a
	// hostname (an IPv6 address in brackets, "[::1]:1883"). The port is
	// required.
	Address string

	// TLS connects with TLS rather than plain TCP. It is never inferred from
	// the port.
	TLS bool

	// ProtocolVersion is MQTT31, MQTT311 or MQTT5. Zero means MQTT311, the
	// libmosquitto default.
	ProtocolVersion byte

	// ClientID may be empty only when CleanStart is set (mosquitto_new). For
	// MQTT 3.1, which requires a client id, an empty id is replaced with a
	// random "mosq-..." id as libmosquitto does.
	ClientID string

	// CleanStart is the clean session (3.x) or clean start (5) flag.
	CleanStart bool

	// KeepAlive in seconds; 0 disables keepalive, otherwise it must be at
	// least 5 (mosquitto__connect_init).
	KeepAlive uint16

	// Username and Password are sent when non-empty. MQTT 3.x does not allow
	// a password without a username.
	Username string
	Password []byte

	// Will, if set, is registered with the server in CONNECT.
	Will *Message

	// ConnectProperties are extra MQTT 5 CONNECT properties.
	ConnectProperties *Properties

	// ReceiveMaximum is the number of inbound QoS 1/2 messages the client
	// accepts concurrently, sent in MQTT 5 CONNECT unless ConnectProperties
	// already sets it. Zero means DefaultReceiveMaximum.
	ReceiveMaximum uint16

	// MaxInflight bounds outbound QoS 1/2 messages awaiting acknowledgement
	// (mosquitto_max_inflight_messages_set). When an MQTT 5 server sends
	// Receive Maximum in CONNACK, that applies instead for the connection.
	// Messages beyond the limit are queued. Zero means DefaultMaxInflight.
	MaxInflight uint16

	// ReconnectDelay, ReconnectDelayMax and ReconnectExponential set how
	// long Run waits before connecting to this server again, or to it next
	// after another server (mosquitto_reconnect_delay_set). The n-th
	// consecutive wait before this server (n from 0) is ReconnectDelay*(n+1),
	// or ReconnectDelay*(n+1)*(n+1) with ReconnectExponential, capped at
	// ReconnectDelayMax. A successful CONNACK from this server resets its n.
	// Zero values mean 1s and ReconnectDelay, libmosquitto's defaults: a
	// fixed 1s.
	ReconnectDelay       time.Duration
	ReconnectDelayMax    time.Duration
	ReconnectExponential bool

	// ConnectTimeout bounds each Connect or Run connection attempt to this
	// server, from dialling to CONNACK. Zero means 30s.
	ConnectTimeout time.Duration

	// TLSConfig is used when TLS is set, and setting it without TLS is an
	// error. If nil, a default configuration is used. An empty ServerName
	// is set from the Address host.
	TLSConfig *tls.Config

	// Dial, if set, opens the network connection to Address in place of a
	// plain net.Dialer, for example to go through a proxy or to wrap the
	// connection. It is called with network "tcp" and the context of the
	// connection attempt. With TLS, the handshake runs over the connection
	// it returns. (*net.Dialer).DialContext has this signature.
	Dial func(ctx context.Context, network, address string) (net.Conn, error)
}

Server is one broker in Options.Servers and the settings used to connect to it. Address is the only required field.

type Subscription 

type Subscription struct {
	Topic string // topic filter; may contain + and # wildcards
	QoS   byte

	// MQTT 5 subscription options. MQTT 3.x has none; they are ignored there,
	// as libmosquitto does.
	NoLocal           bool // do not receive this client's own publications
	RetainAsPublished bool // keep the retain flag of forwarded messages
	RetainHandling    byte // 0 send retained messages, 1 only for a new subscription, 2 never
}

Subscription is one topic filter in a SUBSCRIBE.

type UserProperty 

type UserProperty = packets.UserProperty

UserProperty is an MQTT 5 user property key/value pair.

Source Files

View all Source files

Directories

Path Synopsis
cmd
interactiveclient command
Command interactiveclient is a test application driven over MQTT.
 Command interactiveclient is a test application driven over MQTT.
internal/cli
Package cli holds what the manual test tools under cmd share: the connection flags, building a client from them (or from a -config file of servers), connect and disconnect with logging, and output helpers.
 Package cli holds what the manual test tools under cmd share: the connection flags, building a client from them (or from a -config file of servers), connect and disconnect with logging, and output helpers.
internal/clitest
Package clitest is a scripted MQTT 3.1.1 broker for testing the tools under cmd without a real one: it accepts CONNECT, acknowledges PUBLISH at every QoS, grants every SUBSCRIBE and then sends its Messages.
 Package clitest is a scripted MQTT 3.1.1 broker for testing the tools under cmd without a real one: it accepts CONNECT, acknowledges PUBLISH at every QoS, grants every SUBSCRIBE and then sends its Messages.
mqttconnect command
Command mqttconnect is a manual test tool for this client.
 Command mqttconnect is a manual test tool for this client.
mqttfuzz command
Command mqttfuzz drives a broker with MQTT control packets in an order you choose, for hardening and edge-case probing.
 Command mqttfuzz drives a broker with MQTT control packets in an order you choose, for hardening and edge-case probing.
mqttpub command
Command mqttpub is a manual test tool for publishing with this client.
 Command mqttpub is a manual test tool for publishing with this client.
mqttsub command
Command mqttsub is a manual test tool for subscribing with this client.
 Command mqttsub is a manual test tool for subscribing with this client.
Command conformance runs this client against libmosquitto's client test suite (mosquitto/test/lib).
 Command conformance runs this client against libmosquitto's client test suite (mosquitto/test/lib).
Package fuzz drives an MQTT broker with control packets in an order a YAML scenario chooses, for hardening and edge-case probing.
 Package fuzz drives an MQTT broker with control packets in an order a YAML scenario chooses, for hardening and edge-case probing.
internal
pattern
Package pattern generates and checks the deterministic payloads shared by mqttpub -pattern, mqttsub -verify and the fuzz engine.
 Package pattern generates and checks the deterministic payloads shared by mqttpub -pattern, mqttsub -verify and the fuzz engine.

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.