MQTT 5 en Go con autopaho: tutorial de cliente con reconexión automática

autopaho es el gestor de conexiones de github.com/eclipse/paho.golang, el cliente MQTT 5 de Eclipse Paho para Go. Le das una lista de URLs de broker y algunos callbacks; él conecta, reconecta con backoff cuando cae la red y te avisa cada vez que se (re)establece una conexión. En este tutorial construimos con él un subscriber y un publisher completos, añadimos credenciales y TLS, y vemos cómo probar la reconexión.

Si solo necesitas MQTT 3.1.1, el cliente paho.mqtt.golang sigue siendo una buena opción. Elige autopaho cuando quieras funciones de MQTT 5 —reason codes, expiración de sesión, user properties, expiración de mensajes— junto con reconexión automática.

paho vs autopaho

El módulo tiene dos capas y conviene saber a cuál estás llamando:

En la práctica configuras autopaho.ClientConfig, que incluye un paho.ClientConfig para el client ID y los callbacks de mensajes, y construyes los paquetes de petición con el paquete paho.

Instalación

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

Ambos paquetes vienen del mismo módulo, así que basta con un go get. Usa una versión reciente de Go; la librería sigue las versiones actuales.

Un cliente completo

Este programa se suscribe a un topic, publica un mensaje cada segundo y se cierra limpiamente con Ctrl+C. Funciona contra el endpoint público de pruebas broker.connectmq.com; cambia el host por tu propio broker cuando estés listo.

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, suscribiendo")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("fallo al suscribir: %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("error de conexión: %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("recibido %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("error del cliente: %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("desconexión del 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("fallo al publicar: %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

Ejecútalo con go run . y deberías ver una línea de recepción por cada mensaje publicado. Para ver el mismo tráfico desde fuera, abre el cliente MQTT online, conéctate a broker.connectmq.com y suscríbete a testmqtt/go-autopaho/#.

Algunos detalles importan:

• NewConnection vuelve de inmediato. Empieza a conectar en segundo plano; AwaitConnection bloquea hasta que la primera conexión tiene éxito o se cancela el contexto.
• Suscríbete en OnConnectionUp. Ese callback se dispara en la primera conexión y en cada reconexión, así que tus suscripciones vuelven solas tras una caída de red.
• OnPublishReceived es un slice. Cada handler devuelve (handled bool, err error); varios handlers pueden inspeccionar el mismo mensaje, por ejemplo uno por prefijo de topic.
• Publish con QoS 1 espera el PUBACK y devuelve un *paho.PublishResponse con el reason code del broker; así informa MQTT 5 de problemas como "not authorized" en una publicación concreta.

Credenciales y TLS

Las versiones recientes de autopaho aceptan las credenciales directamente en la configuración. Para TLS, cambia el esquema de la URL a mqtts:// y el puerto a 8883; la librería estándar de Go verifica el certificado contra las raíces del 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")),
	// Solo necesario para una CA privada o certificados de cliente:
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls es una lista: autopaho prueba las URLs en orden, lo que te da un failover sencillo entre dos endpoints. También acepta URLs ws:// y wss:// (por ejemplo wss://broker.connectmq.com:8084/mqtt) cuando los puertos TCP están bloqueados. Si falla el handshake, la guía para probar conexiones TLS explica cómo revisar la cadena de certificados con openssl s_client.

Sesiones persistentes en lugar de resuscribirse

Suscribirse en OnConnectionUp es simple y robusto. La alternativa es que el broker te recuerde:

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

El endpoint de pruebas compartido no conserva sesiones offline, así que prueba esta parte contra tu propia instancia privada de broker.

Con un client ID fijo, el broker guarda las suscripciones y los mensajes QoS 1/2 en cola hasta una hora después de una desconexión. El campo ack.SessionPresent en OnConnectionUp indica si la sesión sobrevivió, para que puedas saltarte la suscripción en ese caso. La expiración de sesión es una función de MQTT 5; consulta sesiones persistentes para ver cómo la gestionan los brokers.

Para mensajes salientes que deben sobrevivir a una desconexión, autopaho ofrece además PublishViaQueue, que guarda los mensajes en una cola (en memoria por defecto, o en disco con el paquete de cola basado en ficheros) y los envía cuando la conexión vuelve.

Comportamiento de reconexión

autopaho reintenta las conexiones fallidas con una espera entre intentos. Las versiones recientes lo exponen como ReconnectBackoff en la configuración (las antiguas usaban un ConnectRetryDelay fijo); el valor por defecto es razonable para la mayoría de clientes. Si gestionas una flota grande, evita que todos los clientes reconecten en el mismo instante tras reiniciar el broker: añade jitter o un backoff creciente, como se explica en buenas prácticas de reconexión.

Los intentos fallidos aparecen en OnConnectError. Mensajes típicos y su significado:

Busca cualquier código numérico en la referencia de reason codes MQTT.

Probar la reconexión

No esperes a producción para saber si la resuscripción funciona. Una comprobación manual rápida:

1. Ejecuta el programa y confirma que los mensajes fluyen.
2. Corta la red (apaga el Wi-Fi, o sudo ifconfig en0 down / sudo ip link set eth0 down) durante más tiempo que el keep-alive.
3. Observa cómo OnConnectError registra los intentos fallidos.
4. Restablece la red. OnConnectionUp debería registrar "conectado, suscribiendo" y las líneas de recepción deberían volver.

Para pruebas automatizadas, envuelve el ida y vuelta publicar/recibir en go test con un timeout; probar clientes MQTT en Go incluye un helper listo para autopaho.

Observa tu cliente autopaho en el navegador

autopaho da a los programas Go un cliente MQTT 5 preparado para producción con muy poco código: configura URLs y callbacks, suscríbete en OnConnectionUp y revisa los reason codes que te devuelve. Prueba un corte de red una vez, mantén los client IDs únicos y tendrás un cliente que se recupera solo.

Preguntas frecuentes