1 files changed, 93 insertions, 0 deletions

diff --git a/mqtt-controller/mqtt.go b/mqtt-controller/mqtt.go
new file mode 100644
index 0000000..0b2687c
--- /dev/null
+++ b/mqtt-controller/mqtt.go
@@ -0,0 +1,93 @@
+package main
+
+import (
+        "encoding/json"
+        "log"
+
+        "github.com/eclipse/paho.mqtt.golang"
+)
+
+type MessageHandler func(topic string, payload []byte)
+
+func mqttSync(t mqtt.Token) error {
+        if t.Wait() && t.Error() != nil {
+                return t.Error()
+        }
+        return nil
+}
+
+type MQTTBroker struct {
+        Devices  chan Device
+        messages chan mqtt.Message
+        done     chan bool
+
+        client mqtt.Client
+}
+
+func NewMQTTBroker(broker string) (*MQTTBroker, error) {
+        bufferSize := 100
+
+        b := &MQTTBroker{
+                Devices:  make(chan Device, bufferSize),
+                messages: make(chan mqtt.Message, bufferSize),
+                done:     make(chan bool),
+        }
+
+        b.client = mqtt.NewClient(
+                mqtt.NewClientOptions().
+                        AddBroker(broker).
+                        SetDefaultPublishHandler(b.handleMessage).
+                        SetClientID("homekit-controller"))
+
+        if err := mqttSync(b.client.Connect()); err != nil {
+                return nil, err
+        }
+
+        return b, nil
+}
+
+func (b *MQTTBroker) Subscribe(topic string, h MessageHandler) error {
+        var handler mqtt.MessageHandler
+        if h != nil {
+                handler = func(c mqtt.Client, m mqtt.Message) {
+                        h(m.Topic(), m.Payload())
+                }
+        }
+
+        if err := mqttSync(b.client.Subscribe(topic, 0, handler)); err != nil {
+                return err
+        }
+
+        return nil
+}
+
+func (b *MQTTBroker) Publish(topic string, msg interface{}) error {
+        if err := mqttSync(b.client.Publish(topic, 0, false, msg)); err != nil {
+                return err
+        }
+        return nil
+}
+
+// Runs synchronously in the mqtt client comms handler goroutine
+func (b *MQTTBroker) handleMessage(c mqtt.Client, m mqtt.Message) {
+        switch m.Topic() {
+        case "shellies/announce", "ubiquiti/announce":
+                a := DeviceAnnounce{}
+                if err := json.Unmarshal(m.Payload(), &a); err != nil {
+                        log.Printf("Unable to unmarshal device announce JSON: %e", err)
+                        return
+                }
+
+                d, err := NewDevice(b, a)
+                if err != nil {
+                        log.Printf("Error creating device driver: %e", err)
+                        return
+                }
+
+                b.Devices <- d
+        }
+}
+
+func (b *MQTTBroker) Shutdown() {
+        b.client.Disconnect(250)
+}