diff options
Diffstat (limited to 'mqtt.go')
| -rw-r--r-- | mqtt.go | 44 |
1 files changed, 29 insertions, 15 deletions
@@ -10,7 +10,9 @@ import ( ) const ( - QOS = byte(1) + QoS0 = byte(0) + QoS1 = byte(1) + QoS2 = byte(2) RETAINED = true MQTT_CONNECT_TIMEOUT = 1 * time.Second MQTT_SUBSCRIBE_TIMEOUT = 1 * time.Second @@ -27,11 +29,21 @@ type MQTTMessage struct { Payload []byte } +type Route struct { + Topic string + QoS byte + Destination chan MQTTMessage +} + +func NewRoute(topic string, qos byte) Route { + return Route{topic, qos, make(chan MQTTMessage)} +} + func (m MQTTMessage) String() string { return fmt.Sprintf("topic='%s' message='%s'", m.Topic, string(m.Payload)) } -func MQTTRun(config MQTTConfig, rx chan MQTTMessage, tx chan MQTTMessage) { +func MQTTRun(config MQTTConfig, tx chan MQTTMessage, routes ...Route) { mqttServerHealthTopic = fmt.Sprintf("%s/server/health", config.TopicPrefix) opts := mqtt.NewClientOptions() @@ -40,19 +52,21 @@ func MQTTRun(config MQTTConfig, rx chan MQTTMessage, tx chan MQTTMessage) { opts.SetCleanSession(true) opts.SetOnConnectHandler(func(c mqtt.Client) { log.Printf("Connected to MQTT broker.") - c.Publish(mqttServerHealthTopic, QOS, true, []byte(`good`)) + c.Publish(mqttServerHealthTopic, QoS1, true, []byte(`good`)) - topic := fmt.Sprintf("%s/cover/+/movement", config.TopicPrefix) - token := c.Subscribe(topic, byte(2), func(c mqtt.Client, msg mqtt.Message) { - message := MQTTMessage{ - Topic: strings.TrimPrefix(msg.Topic(), config.TopicPrefix + "/"), - Payload: msg.Payload(), + for _, route := range routes { + topic := config.TopicPrefix + "/" + route.Topic + token := c.Subscribe(topic, route.QoS, func(c mqtt.Client, msg mqtt.Message) { + message := MQTTMessage{ + Topic: strings.TrimPrefix(msg.Topic(), config.TopicPrefix + "/"), + Payload: msg.Payload(), + } + route.Destination <- message + }) + success := token.WaitTimeout(MQTT_SUBSCRIBE_TIMEOUT) + if !success { + log.Fatalf("Topic subscription failed for topic '%s'", topic) } - rx <- message - }) - success := token.WaitTimeout(MQTT_SUBSCRIBE_TIMEOUT) - if !success { - log.Fatal("Initial topic subscription failed.") } }) opts.SetConnectionLostHandler(MQTTConnectionLostHandler) @@ -60,7 +74,7 @@ func MQTTRun(config MQTTConfig, rx chan MQTTMessage, tx chan MQTTMessage) { opts.SetConnectRetry(true) opts.SetConnectTimeout(MQTT_CONNECT_TIMEOUT) opts.SetKeepAlive(MQTT_KEEPALIVE_PERIOD) - opts.SetWill(mqttServerHealthTopic, `bad`, QOS, true) + opts.SetWill(mqttServerHealthTopic, `bad`, QoS1, true) client := mqtt.NewClient(opts) @@ -74,7 +88,7 @@ func MQTTRun(config MQTTConfig, rx chan MQTTMessage, tx chan MQTTMessage) { for message := range tx { topic := fmt.Sprintf("%s/%s", config.TopicPrefix, message.Topic) - client.Publish(topic, QOS, RETAINED, message.Payload) + client.Publish(topic, QoS1, RETAINED, message.Payload) } } |