MQTT 5 in Go with autopaho: Auto-Reconnect Client Tutorial

autopaho is the connection manager in github.com/eclipse/paho.golang, the Eclipse Paho MQTT 5 client for Go. You give it a list of broker URLs and a few callbacks; it connects, reconnects with backoff when the network drops, and calls you back every time a connection is (re)established. This tutorial builds a complete subscriber and publisher with it, adds credentials and TLS, and shows how to test reconnection.

If you only need MQTT 3.1.1, the older paho.mqtt.golang client is still fine. Choose autopaho when you want MQTT 5 features — reason codes, session expiry, user properties, message expiry — together with automatic reconnection.

paho vs autopaho

The module contains two layers, and it helps to know which one you are calling:

In practice you configure autopaho.ClientConfig, which embeds a paho.ClientConfig for the client ID and message callbacks, and you build request packets from the paho package.

Install

mkdir autopaho-demo && cd autopaho-demo
go mod init example.com/autopaho-demo
go get github.com/eclipse/paho.golang@latest

Both packages come from the same module, so one go get is enough. Use a recent Go release; the library follows current Go versions.

A complete client

This program subscribes to a topic, publishes a message every second and shuts down cleanly on Ctrl+C. It runs against the public test endpoint broker.connectmq.com; replace the host with your own broker when you are ready.

package main

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

const topic = "testmqtt/go-autopaho/demo"

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	u, err := url.Parse("mqtt://broker.connectmq.com:1883")
	if err != nil {
		log.Fatal(err)
	}

	cfg := autopaho.ClientConfig{
		ServerUrls:                    []*url.URL{u},
		KeepAlive:                     30,
		CleanStartOnInitialConnection: true,
		SessionExpiryInterval:         0,
		OnConnectionUp: func(cm *autopaho.ConnectionManager, ack *paho.Connack) {
			log.Println("connected, subscribing")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("subscribe failed: %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("connect error: %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("received %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("client error: %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("server disconnect, reason code %d", d.ReasonCode)
			},
		},
	}

	cm, err := autopaho.NewConnection(ctx, cfg)
	if err != nil {
		log.Fatal(err)
	}
	if err := cm.AwaitConnection(ctx); err != nil {
		log.Fatal(err)
	}

	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for i := 0; ; i++ {
		select {
		case <-ticker.C:
			_, err := cm.Publish(ctx, &paho.Publish{
				Topic:   topic,
				QoS:     1,
				Payload: []byte(fmt.Sprintf("message %d", i)),
			})
			if err != nil {
				log.Printf("publish failed: %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

Run it with go run . and you should see a received line for every published message. To watch the same traffic from outside, open the online MQTT client, connect to broker.connectmq.com and subscribe to testmqtt/go-autopaho/#.

A few details matter here:

• NewConnection returns immediately. It starts connecting in the background; AwaitConnection blocks until the first connection succeeds or the context is cancelled.
• Subscribe in OnConnectionUp. That callback fires on the first connection and on every reconnection, so your subscriptions come back automatically after a network drop.
• OnPublishReceived is a slice. Each handler returns (handled bool, err error); several handlers can inspect the same message, for example one per topic prefix.
• Publish at QoS 1 waits for the PUBACK and returns a *paho.PublishResponse with the broker's reason code, which is how MQTT 5 reports problems such as "not authorized" on a single publish.

Credentials and TLS

Recent versions of autopaho take credentials directly in the config. For TLS, switch the URL scheme to mqtts:// and the port to 8883; the Go standard library verifies the certificate against the system roots.

u, _ := url.Parse("mqtts://broker.connectmq.com:8883")

cfg := autopaho.ClientConfig{
	ServerUrls:      []*url.URL{u},
	KeepAlive:       30,
	ConnectUsername: os.Getenv("MQTT_USER"),
	ConnectPassword: []byte(os.Getenv("MQTT_PASS")),
	// Only needed for a private CA or client certificates:
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls is a list: autopaho tries the URLs in order, which gives you a simple failover between two endpoints. ws:// and wss:// URLs (for example wss://broker.connectmq.com:8084/mqtt) are also accepted when TCP ports are blocked. If the handshake fails, the TLS connection test guide shows how to check the certificate chain with openssl s_client.

Persistent sessions instead of resubscribing

Subscribing in OnConnectionUp is simple and robust. The alternative is to let the broker remember you:

cfg.CleanStartOnInitialConnection = false
cfg.SessionExpiryInterval = 3600 // seconds

The shared test endpoint does not keep offline sessions, so try this part against your own private broker instance.

With a fixed client ID, the broker keeps subscriptions and queued QoS 1/2 messages for up to an hour after a disconnect. The ack.SessionPresent field in OnConnectionUp tells you whether the session survived, so you can skip the subscribe when it did. Session expiry is an MQTT 5 feature; see persistent sessions for how brokers handle it.

For outgoing messages that must survive a disconnect, autopaho also offers PublishViaQueue, which stores messages in a queue (in memory by default, or on disk with the file-based queue package) and sends them when the connection is up again.

Reconnect behaviour

autopaho retries failed connections with a delay between attempts. Recent releases expose this as ReconnectBackoff on the config (older ones used a fixed ConnectRetryDelay); the default is sensible for most clients. When you run a large fleet, make sure clients do not all reconnect at the same instant after a broker restart — add jitter or a growing backoff, as described in reconnect best practices.

OnConnectError is where failed attempts show up. Typical messages and what they mean:

Look up any numeric code in the MQTT reason code reference.

Testing reconnection

Do not wait for production to find out whether resubscription works. A quick manual check:

1. Run the program and confirm messages flow.
2. Cut the network (turn off Wi-Fi, or sudo ifconfig en0 down / sudo ip link set eth0 down) for longer than the keep-alive.
3. Watch OnConnectError log the failed attempts.
4. Restore the network. OnConnectionUp should log "connected, subscribing" and received lines should resume.

For automated tests, wrap the publish/receive round trip in go test with a timeout — testing MQTT clients in Go has a ready-made autopaho helper.

Watch your autopaho client in the browser

autopaho gives Go programs a production-shaped MQTT 5 client with very little code: configure URLs and callbacks, subscribe in OnConnectionUp, and check the reason codes it hands back. Test a network cut once, keep client IDs unique, and you have a client that recovers on its own.

Frequently asked questions