first commit

mqtt.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+// MQTT represents an MQTT client configuration
+type MQTT struct {
+	Host   string
+	Port   string
+	User   string
+	Pass   string
+	Client mqtt.Client
+	id     string
+}
+
+// Server returns a new MQTT connection string
+func (m *MQTT) Server() string {
+	return fmt.Sprintf("tcp://%s:%s", m.Host, m.Port)
+}
+
+// Connect connects to an MQTT broker
+func (m *MQTT) Connect(id string) error {
+	log.Printf("[MQTT:Connect] Connecting: %s", id)
+
+	m.id = id
+
+	opts := mqtt.NewClientOptions().AddBroker(m.Server())
+	opts.SetClientID(id)
+	opts.SetKeepAlive(30 * time.Second)
+	opts.SetPingTimeout(10 * time.Second)
+	if m.User != "" {
+		opts.SetUsername(m.User)
+	}
+	if m.Pass != "" {
+		opts.SetPassword(m.Pass)
+	}
+
+	log.Print("[MQTT] Creating")
+	m.Client = mqtt.NewClient(opts)
+
+	log.Print("[MQTT] Connecting")
+	if token := m.Client.Connect(); token.Wait() && token.Error() != nil {
+		return (token.Error())
+	}
+	return nil
+}
+
+// Disconnect disconnects from an MQTT broker
+func (m *MQTT) Disconnect() {
+	log.Print("[MQTT] Disconnecting")
+	m.Client.Disconnect(250)
+}
+
+// Publish publishes a message to an MQTT topic
+func (m *MQTT) Publish(name string, r *Reading) {
+	log.Printf("[MQTT:Publish] %s (%s), %s", name, r.String(), m.id)
+	format := "sensors/%s/%s/%s"
+
+	// Publish Temperature
+	{
+		topic := fmt.Sprintf(format, m.id, name, "temperature")
+		value := fmt.Sprintf("%04f", r.Temperature)
+		m.publish(topic, value)
+	}
+	// Publish Humidity
+	{
+		topic := fmt.Sprintf(format, m.id, name, "humidity")
+		value := fmt.Sprintf("%04f", r.Humidity)
+		m.publish(topic, value)
+	}
+}
+func (m *MQTT) publish(topic string, payload interface{}) {
+	log.Printf("[MQTT:publish] %s (%v)", topic, payload)
+	m.Client.Publish(topic, 0, false, payload)
+}
+