openpdu/src/mqtt/mqtt.go

package mqtt

import (
	"fmt"
	"time"

	"git.openpdu.org/OpenPDU/openpdu/syslog"
	MQTT "github.com/eclipse/paho.mqtt.golang"
	"github.com/spf13/viper"
)

var MQTTclient MQTT.Client

func init() {
	viper.SetDefault("Mqtt.CliendID", "OpenPDU") // max 23 chars
	viper.SetDefault("Mqtt.Prefix", "openpdu")   // max 23 chars
	viper.SetDefault("Mqtt.Schema", "tcp")
	viper.SetDefault("Mqtt.Host", "localhost")
	viper.SetDefault("Mqtt.Port", "1883")
	viper.SetDefault("Mqtt.Username", "")
	viper.SetDefault("Mqtt.Password", "")
	viper.SetDefault("Mqtt.HomeAssistant", false)

	// MQTT.ERROR = log.New(os.Stdout, "[ERROR] ", 0)
	// MQTT.CRITICAL = log.New(os.Stdout, "[CRIT] ", 0)
	// MQTT.WARN = log.New(os.Stdout, "[WARN]  ", 0)
	// MQTT.DEBUG = log.New(os.Stdout, "[DEBUG] ", 0)

}

// https://girishjoshi.io/post/golang-paho-mqtt/

func Setup() {
	uri := viper.GetString("Mqtt.Schema") + "://" + viper.GetString("Mqtt.Host") + ":" + viper.GetString("Mqtt.Port")
	opts := MQTT.NewClientOptions().AddBroker(uri)

	opts.SetClientID(viper.GetString("Mqtt.CliendID"))

	if username := viper.GetString("Mqtt.Username"); username != "" {
		opts.SetUsername(username)
	}

	if password := viper.GetString("Mqtt.Password"); password != "" {
		opts.SetPassword(password)
	}

	opts.SetAutoReconnect(true)
	opts.SetConnectRetryInterval(5 * time.Second)
	opts.SetConnectTimeout(5 * time.Second)

	opts.SetConnectionLostHandler(func(c MQTT.Client, err error) {
		syslog.Err("mqtt connection lost error: " + err.Error())
	})
	opts.SetReconnectingHandler(func(c MQTT.Client, options *MQTT.ClientOptions) {
		syslog.Notice("mqtt reconnecting")
	})
	opts.SetOnConnectHandler(func(c MQTT.Client) {
		syslog.Notice("mqtt connected")
		//	MQTTclient.Publish("openpdu/status", 0, false, "connected")
	})

	MQTTclient = MQTT.NewClient(opts)

	for {
		token := MQTTclient.Connect()
		token.Wait()
		if MQTTclient.IsConnected() {
			break
		}
		time.Sleep(5 * time.Second)
	}

}

func Subscribe(topic string, handler func(MQTT.Client, MQTT.Message)) {
	// MQTTHandler(MQTT.Client, MQTT.Message)
	MQTTclient.Subscribe(viper.GetString("Mqtt.Prefix")+"/switch/"+topic, 1, handler)
	syslog.Debug(fmt.Sprintf("MQTT subscribed to topic: %s", topic))
}

func Connected() bool {
	if MQTTclient == nil {
		return false
	}
	return MQTTclient.IsConnected()
}

func Disconnect() {
	MQTTclient.Disconnect(250)
}

func Publish(topic string, value string) {
	if MQTTclient == nil {
		return
	}
	if MQTTclient.IsConnected() {
		MQTTclient.Publish(viper.GetString("Mqtt.Prefix")+"/switch/"+topic, 0, false, value)
	}
}

func PublishRoot(topic string, value string) {
	if MQTTclient == nil {
		return
	}
	if MQTTclient.IsConnected() {
		MQTTclient.Publish(topic, 0, false, value)
	}
}