/
goOpenhab_mqtt.go
72 lines (62 loc) · 2.19 KB
/
goOpenhab_mqtt.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
package main
import (
"fmt"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func publishMqtt(mess chan Mqttparms) {
broker := genVar.Mqttbroker
var topic string
var message string
var clientId = "go_mqtt_client"
opts := mqtt.NewClientOptions()
opts.AddBroker(broker)
opts.SetClientID(clientId)
opts.SetDefaultPublishHandler(messagePubHandler)
opts.OnConnect = connectHandler
opts.OnConnectionLost = connectLostHandler
client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
traceLog(fmt.Sprintln(token.Error()))
}
qos := 1
// Subscribe to a topics
for _, topic := range topics {
if token := client.Subscribe(topic, byte(qos), nil); token.Wait() && token.Error() != nil {
traceLog(fmt.Sprintln(token.Error()))
}
}
for {
// Publish a message
inmsg := <-mess
topic = inmsg.Topic
message = inmsg.Message
token := client.Publish(topic, byte(qos), false, message)
token.Wait()
debugLog(5, fmt.Sprintf("Message published to topic %s: %s", topic, message))
time.Sleep(1 * time.Second)
}
}
var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
debugLog(5, fmt.Sprintf("mqtt message received: %s from topic: %s", msg.Payload(), msg.Topic()))
createMessage("mqtt.pubhandler.event", msg.Topic(), string(msg.Payload()))
}
var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
traceLog(fmt.Sprintln("mqtt connected"))
}
// Modified connectLostHandler with reconnect logic
var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
traceLog(fmt.Sprintf("mqtt connection lost: %v", err))
traceLog("mqtt attempting to reconnect...")
for {
time.Sleep(5 * time.Second) // Wait for 5 seconds before trying to reconnect
if token := client.Connect(); token.Wait() && token.Error() == nil {
traceLog(fmt.Sprintln("mqtt reconnected"))
break // Exit the loop once reconnected
} else {
traceLog(fmt.Sprintf("mqtt reconnect attempt failed: %v", token.Error()))
createMessage("mqtt.reconnect.event", fmt.Sprintf("%v", token.Error()), "")
// You may choose to implement additional logic to limit the number of retries or to handle failures differently
}
}
}