Initial commit

Config/Config.go

@@ -0,0 +1,32 @@
+package Config
+
+import (
+	//"fmt"
+	"log"
+
+	"github.com/pelletier/go-toml"
+)
+
+var BrokerAddress = ""
+var BrokerPort int64 = 8883
+var BrokerClientId = "go_client"
+var BrokerUsername = ""
+var BrokerPassword = ""
+var EnableTLS = true
+
+func ReadConfig() {
+	config, err := toml.LoadFile("config.toml")
+
+	if err != nil {
+		//fmt.Println("Error ", err.Error())
+		log.Fatal(err.Error())
+	} else {
+		// retrieve data directly
+		BrokerAddress = config.Get("BROKER.Address").(string)
+		BrokerPort = config.Get("BROKER.Port").(int64)
+		BrokerClientId = config.Get("BROKER.ClientId").(string)
+		BrokerUsername = config.Get("BROKER.Username").(string)
+		BrokerPassword = config.Get("BROKER.Password").(string)
+		EnableTLS = config.Get("BROKER.EnableTLS").(bool)
+	}
+}

main.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"mqttListener/Config"
+	"mqttListener/mqtt"
+
+	//"mqttListener/cmd"
+
+	//"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+
+	Config.ReadConfig()
+
+	// DEBUG Config
+	// fmt.Println("BrokerAddress = ", Config.BrokerAddress)
+
+	mqtt.Setup()
+	mqtt.Connect()
+	go mqtt.Listen()
+
+	//cmd.Exec()
+
+	<-c // wait for SIGTERM
+
+	mqtt.Disconnect()
+}

mqtt/mqtt.go

@@ -0,0 +1,111 @@
+package mqtt
+
+import (
+	"crypto/tls"
+	//"crypto/x509"
+	"fmt"
+	"log"
+	"net/url"
+	"strconv"
+
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	//"os"
+	"mqttListener/Config"
+	"time"
+)
+
+/*
+ URI = mqtt://<user>:<pass>@<server>:<port>
+*/
+
+/* TODO: This options are just a suggestion
+Options:
+ [-help]                      Display help
+ [-customer]                  Customer- or project number
+ [-device]                    Device number
+ [-uri <uri>]                 Broker URI
+*/
+
+var client MQTT.Client
+var opts *MQTT.ClientOptions
+var topic string
+var uri url.URL
+
+func Connect() {
+	client = MQTT.NewClient(opts)
+	token := client.Connect()
+	for !token.WaitTimeout(3 * time.Second) {
+	}
+	if err := token.Error(); err != nil {
+		log.Fatal(err)
+		return
+	}
+	fmt.Println("connected to broker, topic = ", topic)
+}
+
+func Listen() {
+	if !client.IsConnected() {
+		log.Fatal("Client is not connected")
+		return
+	}
+	client.Subscribe(topic, 0, func(client MQTT.Client, msg MQTT.Message) {
+		fmt.Printf("---------------------------------------------------\n")
+		fmt.Printf("TOPIC: %s\n", msg.Topic())
+		fmt.Printf("MSG: \n%s\n", msg.Payload())
+		//fmt.Printf("* [%s] %s\n", msg.Topic(), string(msg.Payload()))
+	})
+}
+
+func Setup() {
+
+	scheme := "tcp://"
+	if Config.EnableTLS {
+		scheme = "ssl://"
+	}
+
+	urlString := scheme
+	urlString += Config.BrokerUsername + ":" + Config.BrokerPassword + "@"
+	urlString += Config.BrokerAddress + ":"
+	urlString += strconv.FormatInt(Config.BrokerPort, 10)
+
+	fmt.Println("broker urlString: ", urlString)
+
+	uri, err := url.Parse(urlString)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	fmt.Println("broker URI: ", *uri)
+	//fmt.Println("customer: ", *customer)
+	//fmt.Println("device: ", *device)
+
+	topic = "/ATB/#"
+
+	opts = createClientOptions(Config.BrokerClientId, uri)
+
+}
+
+func createClientOptions(ClientID string, uri *url.URL) *MQTT.ClientOptions {
+	opts := MQTT.NewClientOptions()
+	opts.AddBroker(uri.Scheme + "://" + uri.Host)
+	opts.SetUsername(uri.User.Username())
+	password, _ := uri.User.Password()
+	opts.SetPassword(password)
+	opts.SetClientID(ClientID)
+	opts.SetDefaultPublishHandler(f)
+	if Config.EnableTLS {
+		tlsConfig := &tls.Config{InsecureSkipVerify: true, ClientAuth: tls.NoClientCert}
+		opts.SetTLSConfig(tlsConfig)
+	}
+
+	return opts
+}
+
+func Disconnect() {
+	client.Disconnect(250)
+}
+
+var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
+	fmt.Printf("TOPIC: %s\n", msg.Topic())
+	fmt.Printf("MSG: %s\n", msg.Payload())
+}