mqtt_client.go

package mqtt

import (
	"fmt"
	"github.com/HummelRummel/Hive/pkg/logger"
	pahomqtt "github.com/eclipse/paho.mqtt.golang"
	"math/rand"
	"path/filepath"
)

type (
	DisconnectCallback = func()
	MessageCallback    = func(topicEp string, msg string)
	topicCallbacks     struct {
		topicBase           string
		disconnectCallbacks []DisconnectCallback
		messageCallbacks    []MessageCallback
	}

	Client struct {
		client               pahomqtt.Client
		host                 string
		callbacksPerTopic    []*topicCallbacks
		statusChangeCallback func(topicBase string, newStatus string)
	}
)

func NewMQTTClient(host string, statusChangeCallback func(topicBase string, newStatus string)) *Client {
	o := &Client{
		host: host,
	}
	opts := pahomqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s", o.host))
	opts.SetClientID(fmt.Sprintf("hive-%d", rand.Int()))
	//opts.SetUsername("emqx")
	//opts.SetPassword("public")
	opts.SetDefaultPublishHandler(o.messagePubHandler)
	opts.OnConnect = o.connectHandler
	opts.OnConnectionLost = o.disconnectHandler
	o.client = pahomqtt.NewClient(opts)
	o.statusChangeCallback = statusChangeCallback
	return o
}

func (o *Client) Run() error {
	logger.Info("trying to connect to MQTT broker [%s]", o.host)
	if token := o.client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

func (o *Client) Publish(topic string, data string) {
	o.client.Publish(topic, 1, false, data)
}

func (o *Client) RegisterCallbacks(topicBase string, messageCallback MessageCallback) {
	for _, cpt := range o.callbacksPerTopic {
		if cpt.topicBase == topicBase {
			cpt.messageCallbacks = append(cpt.messageCallbacks, messageCallback)
			return
		}
	}
	o.callbacksPerTopic = append(o.callbacksPerTopic, &topicCallbacks{
		topicBase:        topicBase,
		messageCallbacks: []MessageCallback{messageCallback},
	})
}

func (o *Client) messagePubHandler(_ pahomqtt.Client, msg pahomqtt.Message) {
}

func (o *Client) messageSubHandler(_ pahomqtt.Client, msg pahomqtt.Message) {
	topicBase := filepath.Dir(msg.Topic())
	topicEp := filepath.Base(msg.Topic())
	data := string(msg.Payload())
	if topicEp == "status" {
		logger.Debug("[%s] changes status to [%s]\n", topicBase, data)
		o.statusChangeCallback(topicBase, data)
		return
	}
	for _, cpt := range o.callbacksPerTopic {
		if cpt.topicBase != topicBase {
			continue
		}
		for _, c := range cpt.messageCallbacks {
			c(topicEp, string(msg.Payload()))
		}
		return
	}

	logger.Debug("no callbacks for topic %s registered: dropping message: %s", msg.Topic(), msg.Payload())
}

func (o *Client) connectHandler(_ pahomqtt.Client) {
	o.client.Subscribe("#", 1, o.messageSubHandler)
	logger.Info("connected to MQTT broker")
}

func (o *Client) disconnectHandler(_ pahomqtt.Client, err error) {
	logger.Info("connection lost to MQTT broker: %s", err)
	for _, cpt := range o.callbacksPerTopic {
		for _, c := range cpt.disconnectCallbacks {
			c()
		}
	}
}