openpdu/src/mqtt.go

package main

import (
	"time"

	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", "")

	// 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)

	//MQTTclient.Subscribe("some_topic", 0, nil);
	//MQTTclient.Disconnect(250)
}

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

func mqttLoop() {
	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) {
		logErr("mqtt connection lost error: " + err.Error())
	})
	opts.SetReconnectingHandler(func(c MQTT.Client, options *MQTT.ClientOptions) {
		logNotice("mqtt reconnecting")
	})
	opts.SetOnConnectHandler(func(c MQTT.Client) {
		logNotice("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)
	}

	for o := range outlets {
		topic := outlets[o].Channel.MQTTCommandTopic()
		if topic == "" {
			continue
		}
		MQTTclient.Subscribe(viper.GetString("Mqtt.Prefix")+"/"+topic, 1, outlets[o].Channel.MQTTHandler)
	}

	go MQTTRefreshLoop()
}

func MQTTreconfigure() {
	if MQTTclient.IsConnected() {
		MQTTclient.Disconnect(250)
	}
	go mqttLoop()
}

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

func MQTTRefreshLoop() {
	for {
		MQTTRefresh()
		time.Sleep(30 * time.Second)
	}
}

func MQTTRefresh() {
	for o := range outlets {
		outlets[o].Channel.UpdateMQTT()
	}
}