-
Notifications
You must be signed in to change notification settings - Fork 0
/
broker.go
103 lines (78 loc) · 2.51 KB
/
broker.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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package daemon
import (
"fmt"
"net/url"
"path/filepath"
"pkg.re/essentialkaos/ek.v10/log"
"strings"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
store "github.com/gongled/dioxy/store"
uuid "github.com/google/uuid"
)
// ////////////////////////////////////////////////////////////////////////////////// //
var datastore *store.Store
// ////////////////////////////////////////////////////////////////////////////////// //
// startObserver starts MQTT listener and updates measurements in memory
func startObserver(ip, port, user, password, topic string, ttl int) error {
maxStoreTime := time.Duration(ttl) * time.Second
datastore = store.NewStore(maxStoreTime)
addr := fmt.Sprintf("tcp://%s:%s@%s:%s/%s", user, password, ip, port, topic)
uri, err := url.Parse(addr)
if err != nil {
log.Crit(err.Error())
shutdown(1)
}
log.Info("Broker listener is connecting to %s:%s", ip, port)
go listenMQTT(uri, topic)
return nil
}
// parseMQTTMessage parses MQTT message to Info struct
func parseMQTTMessage(msg mqtt.Message) *store.Info {
topic := strings.ReplaceAll(msg.Topic(), "/", "@")
metrics := filepath.Base(msg.Topic())
value := string(msg.Payload())
return &store.Info{
Topic: topic,
Metrics: metrics,
Value: value,
}
}
// setMQTTOptions sets up MQTT client to connect as subscriber
func setMQTTOptions(clientId string, uri *url.URL) *mqtt.ClientOptions {
opts := mqtt.NewClientOptions()
password, _ := uri.User.Password()
opts.AddBroker(fmt.Sprintf("tcp://%s", uri.Host))
opts.SetUsername(uri.User.Username())
opts.SetPassword(password)
opts.SetClientID(clientId)
return opts
}
// connectMQTT connects to MQTT broker to operate with metrics
func connectMQTT(clientId string, uri *url.URL) mqtt.Client {
opts := setMQTTOptions(clientId, uri)
client := mqtt.NewClient(opts)
token := client.Connect()
for !token.WaitTimeout(3 * time.Second) {
}
if err := token.Error(); err != nil {
log.Crit("Error while connecting to MQTT broker due to: %s", err.Error())
shutdown(1)
}
return client
}
// listenMQTT listens MQTT broker and update store in memory with latest data
func listenMQTT(uri *url.URL, topic string) {
clientId := uuid.New()
client := connectMQTT(clientId.String(), uri)
token := client.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {
datastore.Add(parseMQTTMessage(msg))
})
for token.Wait() {
time.Sleep(1 * time.Second)
}
if err := token.Error(); err != nil {
log.Crit("Error while operating with MQTT broker due to: %s", err.Error())
shutdown(1)
}
}