MQTT 5 en Go avec autopaho : tutoriel client avec reconnexion automatique

autopaho est le gestionnaire de connexion de github.com/eclipse/paho.golang, le client MQTT 5 d'Eclipse Paho pour Go. Vous lui donnez une liste d'URLs de broker et quelques callbacks ; il se connecte, se reconnecte avec backoff quand le réseau tombe et vous rappelle à chaque fois qu'une connexion est (r)établie. Ce tutoriel construit avec lui un subscriber et un publisher complets, ajoute identifiants et TLS, puis montre comment tester la reconnexion.

Si vous n'avez besoin que de MQTT 3.1.1, l'ancien client paho.mqtt.golang reste un bon choix. Choisissez autopaho quand vous voulez les fonctionnalités de MQTT 5 — reason codes, expiration de session, user properties, expiration des messages — avec la reconnexion automatique.

paho vs autopaho

Le module comporte deux couches, et il est utile de savoir laquelle vous appelez :

En pratique, vous configurez autopaho.ClientConfig, qui embarque un paho.ClientConfig pour le client ID et les callbacks de messages, et vous construisez les paquets de requête avec le package paho.

Installation

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

Les deux packages viennent du même module : un seul go get suffit. Utilisez une version récente de Go ; la bibliothèque suit les versions actuelles.

Un client complet

Ce programme s'abonne à un topic, publie un message chaque seconde et s'arrête proprement avec Ctrl+C. Il tourne contre l'endpoint public de test broker.connectmq.com ; remplacez l'hôte par votre propre broker quand vous êtes prêt.

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("connecté, abonnement en cours")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("échec du subscribe : %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("erreur de connexion : %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("reçu %s : %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("erreur client : %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("déconnexion par le serveur, 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("échec du publish : %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

Lancez-le avec go run . : vous devriez voir une ligne de réception pour chaque message publié. Pour observer le même trafic de l'extérieur, ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et abonnez-vous à testmqtt/go-autopaho/#.

Quelques détails comptent :

• NewConnection rend la main immédiatement. La connexion démarre en arrière-plan ; AwaitConnection bloque jusqu'à ce que la première connexion réussisse ou que le contexte soit annulé.
• Abonnez-vous dans OnConnectionUp. Ce callback se déclenche à la première connexion et à chaque reconnexion : vos abonnements reviennent donc automatiquement après une coupure réseau.
• OnPublishReceived est un slice. Chaque handler renvoie (handled bool, err error) ; plusieurs handlers peuvent examiner le même message, par exemple un par préfixe de topic.
• Publish en QoS 1 attend le PUBACK et renvoie un *paho.PublishResponse avec le reason code du broker — c'est ainsi que MQTT 5 signale des problèmes comme "not authorized" sur une publication précise.

Identifiants et TLS

Les versions récentes d'autopaho acceptent les identifiants directement dans la configuration. Pour TLS, passez le schéma de l'URL en mqtts:// et le port en 8883 ; la bibliothèque standard de Go vérifie le certificat avec les racines du système.

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")),
	// Nécessaire uniquement pour une CA privée ou des certificats client :
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls est une liste : autopaho essaie les URLs dans l'ordre, ce qui offre un failover simple entre deux endpoints. Les URLs ws:// et wss:// (par exemple wss://broker.connectmq.com:8084/mqtt) sont aussi acceptées quand les ports TCP sont bloqués. Si le handshake échoue, le guide de test de connexion TLS montre comment vérifier la chaîne de certificats avec openssl s_client.

Sessions persistantes plutôt que réabonnement

S'abonner dans OnConnectionUp est simple et robuste. L'alternative consiste à laisser le broker se souvenir de vous :

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

L'endpoint de test partagé ne conserve pas les sessions hors ligne : testez donc cette partie avec votre propre instance de broker privée.

Avec un client ID fixe, le broker garde les abonnements et les messages QoS 1/2 en file jusqu'à une heure après une déconnexion. Le champ ack.SessionPresent dans OnConnectionUp indique si la session a survécu, ce qui permet de sauter le subscribe dans ce cas. L'expiration de session est une fonctionnalité de MQTT 5 ; voir sessions persistantes pour la façon dont les brokers la gèrent.

Pour les messages sortants qui doivent survivre à une déconnexion, autopaho propose aussi PublishViaQueue, qui stocke les messages dans une file (en mémoire par défaut, ou sur disque avec le package de file basé sur des fichiers) et les envoie quand la connexion revient.

Comportement de reconnexion

autopaho réessaie les connexions échouées avec un délai entre les tentatives. Les versions récentes l'exposent via ReconnectBackoff dans la configuration (les anciennes utilisaient un ConnectRetryDelay fixe) ; la valeur par défaut convient à la plupart des clients. Avec une flotte importante, évitez que tous les clients se reconnectent au même instant après un redémarrage du broker : ajoutez du jitter ou un backoff croissant, comme décrit dans bonnes pratiques de reconnexion.

Les tentatives échouées apparaissent dans OnConnectError. Messages typiques et leur signification :

Retrouvez n'importe quel code numérique dans la référence des reason codes MQTT.

Tester la reconnexion

N'attendez pas la production pour savoir si le réabonnement fonctionne. Une vérification manuelle rapide :

1. Lancez le programme et vérifiez que les messages circulent.
2. Coupez le réseau (désactivez le Wi-Fi, ou sudo ifconfig en0 down / sudo ip link set eth0 down) plus longtemps que le keep-alive.
3. Regardez OnConnectError journaliser les tentatives échouées.
4. Rétablissez le réseau. OnConnectionUp doit afficher "connecté, abonnement en cours" et les lignes de réception doivent reprendre.

Pour des tests automatisés, encapsulez l'aller-retour publication/réception dans go test avec un timeout — tester des clients MQTT en Go propose un helper autopaho prêt à l'emploi.

Observez votre client autopaho dans le navigateur

autopaho offre aux programmes Go un client MQTT 5 prêt pour la production avec très peu de code : configurez URLs et callbacks, abonnez-vous dans OnConnectionUp et vérifiez les reason codes renvoyés. Testez une coupure réseau une fois, gardez des client IDs uniques, et vous aurez un client qui se rétablit tout seul.

Questions fréquentes