MQTT 5 em Go com autopaho: tutorial de cliente com reconexão automática

autopaho é o gerenciador de conexões do github.com/eclipse/paho.golang, o cliente MQTT 5 do Eclipse Paho para Go. Você passa uma lista de URLs de broker e alguns callbacks; ele conecta, reconecta com backoff quando a rede cai e avisa você toda vez que uma conexão é (re)estabelecida. Este tutorial monta um subscriber e um publisher completos com ele, adiciona credenciais e TLS e mostra como testar a reconexão.

Se você só precisa de MQTT 3.1.1, o cliente paho.mqtt.golang continua ótimo. Escolha o autopaho quando quiser recursos do MQTT 5 — reason codes, expiração de sessão, user properties, expiração de mensagens — junto com reconexão automática.

paho vs autopaho

O módulo tem duas camadas, e vale saber qual delas você está chamando:

Na prática você configura autopaho.ClientConfig, que embute um paho.ClientConfig para o client ID e os callbacks de mensagem, e monta os pacotes de requisição com o pacote paho.

Instalação

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

Os dois pacotes vêm do mesmo módulo, então um único go get basta. Use uma versão recente do Go; a biblioteca acompanha as versões atuais.

Um cliente completo

Este programa se inscreve em um topic, publica uma mensagem por segundo e encerra de forma limpa com Ctrl+C. Ele roda contra o endpoint público de testes broker.connectmq.com; troque o host pelo seu próprio broker quando estiver pronto.

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("conectado, fazendo subscribe")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("falha no subscribe: %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("erro de conexão: %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("recebido %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("erro do cliente: %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("desconexão pelo servidor, 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("falha no publish: %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

Rode com go run . e você verá uma linha de recebimento para cada mensagem publicada. Para ver o mesmo tráfego de fora, abra o cliente MQTT online, conecte em broker.connectmq.com e se inscreva em testmqtt/go-autopaho/#.

Alguns detalhes importam:

• NewConnection retorna na hora. Ele começa a conectar em segundo plano; AwaitConnection bloqueia até a primeira conexão dar certo ou o contexto ser cancelado.
• Faça o subscribe no OnConnectionUp. Esse callback dispara na primeira conexão e em cada reconexão, então suas inscrições voltam sozinhas depois de uma queda de rede.
• OnPublishReceived é um slice. Cada handler retorna (handled bool, err error); vários handlers podem inspecionar a mesma mensagem, por exemplo um por prefixo de topic.
• Publish com QoS 1 espera o PUBACK e retorna um *paho.PublishResponse com o reason code do broker — é assim que o MQTT 5 reporta problemas como "not authorized" em um publish específico.

Credenciais e TLS

As versões recentes do autopaho recebem as credenciais direto na configuração. Para TLS, mude o esquema da URL para mqtts:// e a porta para 8883; a biblioteca padrão do Go verifica o certificado contra as raízes do sistema.

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")),
	// Só é necessário para uma CA privada ou certificados de cliente:
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls é uma lista: o autopaho tenta as URLs em ordem, o que dá um failover simples entre dois endpoints. URLs ws:// e wss:// (por exemplo wss://broker.connectmq.com:8084/mqtt) também são aceitas quando as portas TCP estão bloqueadas. Se o handshake falhar, o guia de teste de conexão TLS mostra como verificar a cadeia de certificados com openssl s_client.

Sessões persistentes em vez de nova inscrição

Fazer o subscribe no OnConnectionUp é simples e robusto. A alternativa é deixar o broker lembrar de você:

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

O endpoint de testes compartilhado não guarda sessões offline, então teste esta parte contra a sua própria instância privada de broker.

Com um client ID fixo, o broker guarda as inscrições e as mensagens QoS 1/2 enfileiradas por até uma hora após uma desconexão. O campo ack.SessionPresent no OnConnectionUp informa se a sessão sobreviveu, para você pular o subscribe nesse caso. A expiração de sessão é um recurso do MQTT 5; veja sessões persistentes para entender como os brokers lidam com isso.

Para mensagens de saída que precisam sobreviver a uma desconexão, o autopaho também oferece PublishViaQueue, que guarda as mensagens em uma fila (em memória por padrão, ou em disco com o pacote de fila baseado em arquivos) e as envia quando a conexão volta.

Comportamento de reconexão

O autopaho tenta de novo as conexões que falham, com um intervalo entre as tentativas. Versões recentes expõem isso como ReconnectBackoff na configuração (as antigas usavam um ConnectRetryDelay fixo); o padrão é sensato para a maioria dos clientes. Se você opera uma frota grande, garanta que os clientes não reconectem todos no mesmo instante depois de um restart do broker — adicione jitter ou um backoff crescente, como em boas práticas de reconexão.

As tentativas que falham aparecem no OnConnectError. Mensagens típicas e o que significam:

Consulte qualquer código numérico na referência de reason codes MQTT.

Testando a reconexão

Não espere a produção para descobrir se a nova inscrição funciona. Uma verificação manual rápida:

1. Rode o programa e confirme que as mensagens estão fluindo.
2. Corte a rede (desligue o Wi-Fi, ou sudo ifconfig en0 down / sudo ip link set eth0 down) por mais tempo que o keep-alive.
3. Veja o OnConnectError registrar as tentativas que falham.
4. Restaure a rede. O OnConnectionUp deve registrar "conectado, fazendo subscribe" e as linhas de recebimento devem voltar.

Para testes automatizados, envolva o ciclo publicar/receber em go test com timeout — testando clientes MQTT em Go traz um helper pronto para o autopaho.

Acompanhe seu cliente autopaho no navegador

O autopaho dá aos programas Go um cliente MQTT 5 com cara de produção usando pouquíssimo código: configure URLs e callbacks, faça o subscribe no OnConnectionUp e confira os reason codes que ele devolve. Teste um corte de rede uma vez, mantenha os client IDs únicos e você terá um cliente que se recupera sozinho.

Perguntas frequentes