move package mqtt to package catbus, make more convenient

1 files changed, 38 insertions, 42 deletions

diff --git a/cmd/catbus-actuator-wakeonlan/main.go b/cmd/catbus-actuator-wakeonlan/main.go
index 53da4d9..37efbbe 100644
--- a/cmd/catbus-actuator-wakeonlan/main.go
+++ b/cmd/catbus-actuator-wakeonlan/main.go
@@ -9,9 +9,9 @@ import (
 	"flag"
 	"log"
 
+	"go.eth.moe/catbus-wakeonlan/catbus"
 	"go.eth.moe/catbus-wakeonlan/config"
 	"go.eth.moe/catbus-wakeonlan/logger"
-	"go.eth.moe/catbus-wakeonlan/mqtt"
 	"go.eth.moe/catbus-wakeonlan/wakeonlan"
 )
 
@@ -34,52 +34,48 @@ func main() {
 		log.WithError(err).Fatal("could not parse config file")
 	}
 
-	log.AddField("broker-uri", config.Broker)
+	log.AddField("broker-uri", config.BrokerURI)
 
-	brokerOptions := mqtt.NewClientOptions()
-	brokerOptions.AddBroker(config.Broker)
-	brokerOptions.SetAutoReconnect(true)
-	brokerOptions.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
-		log := log
-		if err != nil {
-			log = log.WithError(err)
-		}
-		log.Error("disconnected from MQTT broker")
-	})
-	brokerOptions.SetOnConnectHandler(func(broker mqtt.Client) {
-		log.Info("connected to MQTT broker")
-
-		for topic := range config.MACsByTopic {
-			token := broker.Subscribe(topic, mqtt.AtLeastOnce, func(_ mqtt.Client, msg mqtt.Message) {
-				if string(msg.Payload()) != "on" {
-					return
-				}
+	catbusOptions := catbus.ClientOptions{
+		DisconnectHandler: func(_ *catbus.Client, err error) {
+			log := log
+			if err != nil {
+				log = log.WithError(err)
+			}
+			log.Error("disconnected from MQTT broker")
+		},
+		ConnectHandler: func(client *catbus.Client) {
+			log.Info("connected to MQTT broker")
 
-				mac, ok := config.MACsByTopic[msg.Topic()]
-				if !ok {
-					return
-				}
+			for topic := range config.MACsByTopic {
+				err := client.Subscribe(topic, func(_ *catbus.Client, msg catbus.Message) {
+					if string(msg.Payload()) != "on" {
+						return
+					}
+					mac, ok := config.MACsByTopic[msg.Topic()]
+					if !ok {
+						return
+					}
 
-				log.AddField("mac", mac)
-				log.AddField("topic", topic)
-				if err := wakeonlan.Wake(mac); err != nil {
-					log.WithError(err).Error("could not send wake-on-lan packet")
-					return
+					log.AddField("mac", mac)
+					log.AddField("topic", topic)
+					if err := wakeonlan.Wake(mac); err != nil {
+						log.WithError(err).Error("could not send wake-on-lan packet")
+						return
+					}
+					log.Info("sent wake-on-lan packet")
+				})
+				if err != nil {
+					log := log.WithError(err)
+					log.AddField("topic", topic)
+					log.Error("could not subscribe to MQTT topic")
 				}
-				log.Info("sent wake-on-lan packet")
-			})
-			if err := token.Error(); err != nil {
-				log := log.WithError(err)
-				log.AddField("topic", topic)
-				log.Error("could not subscribe to MQTT topic")
 			}
-		}
-	})
-
-	broker := mqtt.NewClient(brokerOptions)
-	if token := broker.Connect(); token.Error() != nil {
-		log.WithError(token.Error()).Fatal("could not connect to MQTT broker")
+		},
 	}
+	catbus := catbus.NewClient(config.BrokerURI, catbusOptions)
 
-	select {}
+	if err := catbus.Connect(); err != nil {
+		log.WithError(err).Fatal("could not connect to MQTT broker")
+	}
 }