mqttconn

README

mqttconn

A minimal MQTT v3.1.1 client: Dial, Publish (QoS 0/1), Subscribe, and Close. No third-party dependency — the CONNECT handshake and packet read/write are implemented directly on top of net.Conn, the same way wsconn implements just enough of RFC 6455 rather than depending on a full-featured library.

Deliberately narrow: no QoS 2, no persistent sessions, no offline message queuing, no TLS helpers beyond what net.Dial itself provides. It exists to serve inovelli/mqtt's Zwave JS UI gateway path — publish a config write, subscribe to value-change updates — not to be a general-purpose MQTT library. If you need one, use eclipse/paho.mqtt.golang instead.

Install

go get github.com/ryanjohnsontv/mqttconn

Quick start

client, err := mqttconn.Dial(ctx, "192.168.1.20:1883", mqttconn.Options{ClientID: "myapp"})
if err != nil {
    log.Fatal(err)
}
defer client.Close()

if err := client.Publish(ctx, "zwave/19/112/0/13/set", []byte(`{"value":85}`), 1); err != nil {
    log.Fatal(err)
}

client.Subscribe(ctx, "zwave/+/91/0/scene/+", func(topic string, payload []byte) {
    log.Printf("%s: %s", topic, payload)
})

Options is optional in spirit — the zero value dials with a random client ID, no auth, and a 60s keep-alive. Every session is a clean session (no queued messages or persistent subscriptions survive a reconnect); this package doesn't track broker-side session state at all.

Publish

QoS 0 returns as soon as the packet is written — fire and forget. QoS 1 blocks until the broker's PUBACK arrives or ctx is done, giving you a real delivery guarantee for writes worth confirming. QoS 2 isn't implemented; neither of this package's consumers need its extra round trip.

Subscribe

handler runs in its own goroutine per message, so it's free to block without stalling the read loop or other subscriptions. Topic filters support both MQTT wildcards: + matches exactly one level, # — valid only as the filter's last level — matches that level and everything after it, including nothing further.

Matching happens client-side: the broker already only delivers messages that satisfy what you subscribed to, but with more than one active subscription this package still needs to work out which handler(s) a given incoming topic belongs to.

Design constraints

• QoS 0 and 1 only. No PUBREC/PUBREL/PUBCOMP handshake.
• Clean sessions only. No persistent subscriptions, no queued messages while disconnected.
• No TLS convenience wrapper. Dial over TCP only; wrap your own tls.Config at the net.Conn level if you need it — not worth adding options for a broker this package's consumers run locally, unencrypted.
• 16MiB per-packet cap, generous for JSON value-update payloads, not configurable.

Verifying against a real broker

This package's own tests run against a hand-rolled fake broker (no real MQTT server involved, the same testing philosophy wsconn uses) — but the protocol itself was also manually verified end to end against a real Eclipse Mosquitto broker (docker run eclipse-mosquitto:2) before shipping: CONNECT/CONNACK, Publish at both QoS levels, and Subscribe correctly receiving a published message back. If you change the wire-level packet encoding, re-verify against a real broker too — a fake broker built by the same hand that wrote the client can share the same misunderstanding of the spec.

Feedback

Issues, questions, and PRs are welcome — this is early, and real-world feedback (what broke, what's confusing, what's missing) is genuinely useful.

Documentation

Overview

Package mqttconn is a minimal MQTT v3.1.1 client: Dial, Publish (QoS 0/1), Subscribe, and Close. No third-party dependency — the CONNECT handshake and packet read/write are implemented directly on top of net.Conn, mirroring how wsconn implements just enough of RFC 6455 rather than depending on a full-featured library.

Deliberately narrow: no QoS 2, no persistent sessions, no offline message queuing, no TLS helpers beyond what net.Dial itself provides. It exists to serve inovelli/mqtt's Zwave JS UI gateway path, whose needs are exactly "publish a config write" and "subscribe to value-change updates" — not to be a general-purpose MQTT library.

Index

• type Client
  • func Dial(ctx context.Context, addr string, opts Options) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Publish(ctx context.Context, topic string, payload []byte, qos byte) error
  • func (c *Client) Subscribe(ctx context.Context, filter string, handler func(topic string, payload []byte)) error
• type Options

Types

type Client

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

Client is one connected MQTT v3.1.1 session. A single goroutine may call Publish/Subscribe concurrently with each other and with Close; only one background goroutine ever reads from the connection.

func Dial

func Dial(ctx context.Context, addr string, opts Options) (*Client, error)

Dial connects to an MQTT v3.1.1 broker at addr ("host:port"), completes the CONNECT/CONNACK handshake, and starts the background read and keep-alive ping loops.

func (*Client) Close

func (c *Client) Close() error

Close sends DISCONNECT and closes the underlying connection, unblocking the read loop and every in-flight Publish/Subscribe call. Safe to call more than once, or concurrently with the read loop noticing the connection died on its own — whichever happens first wins the close error; cleanup always happens.

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, topic string, payload []byte, qos byte) error

Publish sends payload to topic at qos (0 or 1 — QoS 2 isn't implemented, this package has no consumer that needs its exactly-once handshake). QoS 1 blocks until the broker's PUBACK arrives or ctx is done; QoS 0 returns as soon as the packet is written.

func (*Client) Subscribe

func (c *Client) Subscribe(ctx context.Context, filter string, handler func(topic string, payload []byte)) error

Subscribe registers handler for every message published to a topic matching filter ("+" for one level, "#" for the rest, per MQTT's topic matching rules), and blocks until the broker's SUBACK arrives or ctx is done. handler is called from its own goroutine per message, so it may block without stalling the connection's read loop.

type Options

type Options struct {
	ClientID  string
	Username  string
	Password  string
	KeepAlive time.Duration // default 60s
}

Options configures Dial. The zero value is a reasonable default: a random client ID, no auth, 60s keep-alive. Every session is a clean session (no persistent subscriptions or queued messages across reconnects) — this package has no use for offline delivery guarantees.