MSE-PI-E2EEDA-Plein-de-eeeeeeeeeee/db/src/mqtt/mqtt.go

// Package mqtt_gateway provides an abstraction to an MQTT broker client.
package mqtt

import (
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	dp "gateway/point"
	"log"
	"strings"
	"time"

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

// A MqttParams is the abstracted MQTT gateway.
// It provides the MQTT parameters to initialize the connection and the method to add data.
type MqttParams struct {
	Broker           string
	ClientId         string
	Qos              byte
	Username         string
	Password         string
	TlsConfig        *tls.Config
	OnConnect        mqtt.OnConnectHandler
	OnConnectionLost mqtt.ConnectionLostHandler
	Timeout          time.Duration
}

// MqttGateway is the abstracted MQTT gateway.
// It connects to the MQTT broker and provides the method to send data to the broker.
type MqttGateway struct {
	MqttParams MqttParams
	Client     mqtt.Client
}

// mqttPayload is the JSON structure published to the broker in a specific topic.
// It contains the values and the timestamp of the data.
type mqttPayload map[string]any

// connectHandler is called when the client connects to the broker. It prints a message to the console.
var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
	log.Println("[MQTT Gateway] Connected to MQTT Broker")
}

// connectLostHandler is called when the client loses connection to the broker. It prints a message to the console with the error.
var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	log.Printf("[MQTT Gateway] Connection lost: %v\n", err)
}

func getTopic(t []dp.Topic) string {
	var topic []string
	for _, t := range t {
		topic = append(topic, t.Content)
	}
	return strings.Join(topic, "/")
}

// NewMqttGateway creates a new MqttGateway with the given parameters.
// And establishes the connection
func NewMqttGateway(p MqttParams) (*MqttGateway, error) {
	// Verify input variable
	if p.Broker == "" {
		return nil, errors.New("[MQTT Gateway] Invalid broker address")
	}

	if p.ClientId == "" {
		return nil, errors.New("[MQTT Gateway] Invalid client id")
	}

	if p.Qos > 2 {
		return nil, errors.New("[MQTT Gateway] Invalid QoS level")
	}

	if p.Timeout == 0 {
		// Set to default value
		p.Timeout = time.Second * 5
	}

	opts := mqtt.NewClientOptions()
	opts.AddBroker(p.Broker)
	opts.SetClientID(p.ClientId)

	if p.TlsConfig != nil {
		opts.SetTLSConfig(p.TlsConfig)
	}

	if p.OnConnect != nil {
		opts.SetOnConnectHandler(p.OnConnect)
	} else {
		opts.SetOnConnectHandler(connectHandler)
	}

	if p.OnConnectionLost != nil {
		opts.SetConnectionLostHandler(p.OnConnectionLost)
	} else {
		opts.SetConnectionLostHandler(connectLostHandler)
	}

	if p.Username != "" {
		opts.SetUsername(p.Username)
		opts.SetPassword(p.Password)
	}

	client := mqtt.NewClient(opts)
	token := client.Connect()
	if !token.WaitTimeout(p.Timeout) {
		return nil, fmt.Errorf("[MQTT Gateway] Mqtt connect timed out")
	}
	if err := token.Error(); err != nil {
		return nil, fmt.Errorf("[MQTT Gateway] Mqtt connect failed: %w", err)
	}

	return &MqttGateway{
		MqttParams: p,
		Client:     client,
	}, nil
}

// SendData is used to send data in the MQTT gateway.
// It uses the DataPointInfo interface for abstracting the generic type of the DataPoint
func (g *MqttGateway) SendData(msg dp.DataPointInfo) error {

	topic := getTopic(msg.Tags())
	if topic == "" {
		return errors.New("[MQTT Gateway] Invalid topic")
	}

	payload := mqttPayload{
		"timestamp": msg.Timestamp().Unix(),
	}

	for key, value := range msg.PayloadAsAny() {
		payload[key] = value
	}

	payloadJson, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("[MQTT Gateway] Failed to marshal payload: %w", err)
	}

	token := g.Client.Publish(topic, g.MqttParams.Qos, false, payloadJson)
	if !token.WaitTimeout(g.MqttParams.Timeout) {
		return fmt.Errorf("[MQTT Gateway] Mqtt connect timed out")
	}
	if token.Error() != nil {
		return fmt.Errorf("[MQTT Gateway] Failed to publish message: %w", token.Error())
	}

	return nil
}

// Disconnect is used to disconnect the MQTT gateway from the broker.
// It prints a message to the console when the disconnection is successful.
func (g *MqttGateway) Disconnect() {
	g.Client.Disconnect(0)
	log.Println("[MQTT Gateway] Disconnected from MQTT Broker")
}

// Subscribe is used to subscribe to a topic in the MQTT gateway.
// It takes a topic and a callback function as parameters.
// The callback function is called when a message is received on the subscribed topic.
func (g *MqttGateway) Subscribe(topic string, callback mqtt.MessageHandler) error {
	token := g.Client.Subscribe(topic, g.MqttParams.Qos, callback)
	if !token.WaitTimeout(g.MqttParams.Timeout) {
		return fmt.Errorf("[MQTT Gateway] MQTT gateway timed out")
	}
	if token.Error() != nil {
		return fmt.Errorf("[MQTT Gateway] MQTT gateway failed to subscribe: %w", token.Error())
	}

	log.Printf("[MQTT Gateway] Subscribed to topic: %s\n", topic)
	return nil
}

// Unsubscribe is used to unsubscribe from a topic in the MQTT gateway.
func (g *MqttGateway) Unsubscribe(topic string) error {
	token := g.Client.Unsubscribe(topic)
	if !token.WaitTimeout(g.MqttParams.Timeout) {
		return fmt.Errorf("[MQTT Gateway] MQTT gateway timed out")
	}
	if token.Error() != nil {
		return fmt.Errorf("[MQTT Gateway] MQTT gateway failed to unsubscribe: %w", token.Error())
	}

	log.Printf("[MQTT Gateway] Unsubscribed from topic: %s\n", topic)
	return nil
}