-
Notifications
You must be signed in to change notification settings - Fork 65
/
mqttClient.go
32 lines (28 loc) · 1013 Bytes
/
mqttClient.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
package mqttclient
import (
mqtt "github.com/eclipse/paho.mqtt.golang"
"go.uber.org/zap"
)
type SubscribeOptions struct {
Topic string
QoS byte
OnMessageReceived mqtt.MessageHandler
Logger *zap.Logger
}
func Subscribe(connectionOptions *mqtt.ClientOptions, subscribeOptions SubscribeOptions) error {
oldConnect := connectionOptions.OnConnect
connectionOptions.OnConnect = func(client mqtt.Client) {
logger := subscribeOptions.Logger
oldConnect(client)
logger.Info("Connected to MQTT Broker")
logger.Info("Will subscribe to topic", zap.String("topic", subscribeOptions.Topic))
if token := client.Subscribe(subscribeOptions.Topic, subscribeOptions.QoS, subscribeOptions.OnMessageReceived); token.Wait() && token.Error() != nil {
logger.Error("Could not subscribe", zap.Error(token.Error()))
}
}
client := mqtt.NewClient(connectionOptions)
if token := client.Connect(); token.Wait() && token.Error() != nil {
return token.Error()
}
return nil
}