nano_metrics/server/transport/mqtt.go

package transport

import (
	"encoding/json"
	"fmt"
	"log"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/user/nanometrics/server/models"
)

type MQTTClient struct {
	client    mqtt.Client
	topicBase string
}

func StartMQTT(broker, topicBase string, handler func(*models.AgentMetrics)) (*MQTTClient, error) {
	mc := &MQTTClient{topicBase: topicBase}

	opts := mqtt.NewClientOptions().
		AddBroker(broker).
		SetClientID("nanometrics-server").
		SetAutoReconnect(true).
		SetOnConnectHandler(func(c mqtt.Client) {
			log.Printf("[mqtt] connecté à %s", broker)
			topic := fmt.Sprintf("%s/+/metrics", topicBase)
			if tok := c.Subscribe(topic, 0, nil); tok.Wait() && tok.Error() != nil {
				log.Printf("[mqtt] subscribe error: %v", tok.Error())
			}
		}).
		SetConnectionLostHandler(func(c mqtt.Client, err error) {
			log.Printf("[mqtt] connexion perdue: %v", err)
		}).
		SetDefaultPublishHandler(func(_ mqtt.Client, msg mqtt.Message) {
			var m models.AgentMetrics
			if err := json.Unmarshal(msg.Payload(), &m); err != nil {
				log.Printf("[mqtt] JSON invalide sur %s: %v", msg.Topic(), err)
				return
			}
			if m.Hostname != "" {
				handler(&m)
			}
		})

	mc.client = mqtt.NewClient(opts)
	if tok := mc.client.Connect(); tok.Wait() && tok.Error() != nil {
		return nil, fmt.Errorf("mqtt connect: %w", tok.Error())
	}
	return mc, nil
}

func (mc *MQTTClient) PushConfig(hostname string, cfg *models.AgentConfig) error {
	topic := fmt.Sprintf("%s/%s/config", mc.topicBase, hostname)
	data, err := json.Marshal(cfg)
	if err != nil {
		return err
	}
	tok := mc.client.Publish(topic, 1, false, data)
	tok.Wait()
	return tok.Error()
}