mqtt initial

go.mod

@@ -5,6 +5,7 @@ go 1.18
 require (
 	github.com/cortesi/devd v0.0.0-20200427000907-c1a3bfba27d8
 	github.com/cortesi/modd v0.0.0-20211215124449-6083f9d1c171
+	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/mdomke/git-semver/v6 v6.2.0
 	github.com/rs/zerolog v1.26.1
 	github.com/spf13/cobra v1.4.0

go.sum

@@ -93,6 +93,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
 github.com/emirpasic/gods v1.12.0/go.mod h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=

screen/listener.go

@@ -0,0 +1,47 @@
+package screen
+
+import (
+	"strings"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+func runListener(cfg configObj) error {
+	log := cfg.logger.With().
+		Str("broker", cfg.brokerUrl).
+		Str("clientid", cfg.clientId).
+		Logger()
+	opts := mqtt.NewClientOptions().
+		AddBroker(cfg.brokerUrl).
+		SetClientID(cfg.clientId).
+		SetDefaultPublishHandler(messageHandler)
+
+	opts = opts.SetOnConnectHandler(func(client mqtt.Client) {
+		log.Info().Str("topic", cfg.topic).Msg("Subscribing to topic")
+		if token := client.Subscribe(cfg.topic, 1, nil); token.Wait() && token.Error() != nil {
+			log.Error().
+				Str("topic", cfg.topic).
+				Err(token.Error()).Msg("Couldn't subscribe to topic")
+		}
+	})
+
+	log.Info().Msg("Connecting to broker")
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Error().Err(token.Error()).Msg("Couldn't connect to broker")
+		return token.Error()
+	}
+	return nil
+}
+
+func messageHandler(client mqtt.Client, msg mqtt.Message) {
+	log := config.logger
+	topic := msg.Topic()
+	payload := msg.Payload()
+	if strings.Compare(string(payload), "\n") > 0 {
+		log.Info().
+			Str("topic", topic).
+			Str("payload", string(payload)).
+			Msg("MQTT Message")
+	}
+}

screen/server.go

@@ -0,0 +1,27 @@
+package screen
+
+import (
+	"github.com/rs/zerolog"
+	"github.com/spf13/viper"
+)
+
+type configObj struct {
+	logger    zerolog.Logger
+	brokerUrl string
+	clientId  string
+	topic     string
+}
+
+var config = configObj{}
+
+func init() {
+	viper.SetDefault("mqtt_client_id", "screenlistener")
+}
+
+func Run(rootLogger zerolog.Logger, exitch chan bool) {
+	config.logger = rootLogger.With().Str("name", "screen").Logger()
+	config.brokerUrl = viper.GetString("mqtt_url")
+	config.clientId = viper.GetString("mqtt_client_id")
+	config.topic = viper.GetString("mqtt_topic")
+	exitch <- true
+}