148 lines
3.8 KiB
Go
148 lines
3.8 KiB
Go
package mqtt
|
|
|
|
import "log"
|
|
import "strings"
|
|
import "fmt"
|
|
import MQTT "github.com/eclipse/paho.mqtt.golang"
|
|
import "github.com/google/uuid"
|
|
import "crypto/tls"
|
|
import "udi/config"
|
|
import "udi/counter"
|
|
|
|
type Message struct {
|
|
Topic string
|
|
Payload []byte
|
|
}
|
|
|
|
var InputChannel chan Message = make(chan Message, 100)
|
|
var OutputChannel chan Message = make(chan Message, 100)
|
|
|
|
var mqttClient MQTT.Client
|
|
|
|
func onMessageReceived(client MQTT.Client, message MQTT.Message) {
|
|
// log.Printf("Message received, topic: %s, payload: %s\n", message.Topic(), message.Payload())
|
|
m := Message {
|
|
Topic: message.Topic(),
|
|
Payload: message.Payload(),
|
|
}
|
|
select {
|
|
case InputChannel <- m:
|
|
counter.S("Received")
|
|
{}
|
|
//log.Println("Message sent to channel")
|
|
default:
|
|
log.Println("Channel full, message lost")
|
|
counter.F("Received")
|
|
}
|
|
}
|
|
|
|
func onConnectionLost(client MQTT.Client, error error) {
|
|
log.Printf("Connection lost, error %s", error)
|
|
}
|
|
|
|
func onReconnecting(client MQTT.Client, clientOpts *MQTT.ClientOptions) {
|
|
log.Println("Oops, connection lost, already reconnecting ...")
|
|
}
|
|
|
|
|
|
func onConnect(client MQTT.Client) {
|
|
var topics []string
|
|
for _, mapping := range config.Config.TopicMappings {
|
|
topics = append(topics, mapping.Topics...)
|
|
}
|
|
if len(topics) == 0 {
|
|
log.Fatal("No topics given, set in configuration")
|
|
}
|
|
for _, topic := range topics {
|
|
if token := client.Subscribe(topic, 0, onMessageReceived); token.Wait() && token.Error() != nil {
|
|
log.Fatalf("Unable to subscribe to topic %s, error %s", topic, token.Error())
|
|
}
|
|
log.Printf("Topic %s subscribed", topic)
|
|
}
|
|
}
|
|
|
|
func outputDispatcher(client MQTT.Client) {
|
|
for {
|
|
select {
|
|
case message := <- OutputChannel:
|
|
log.Printf("Message arrived in outputDispatcher, topic: %s, payload: %s\n", message.Topic, message.Payload)
|
|
if token := client.Publish(message.Topic, 0, false, message.Payload); token.Wait() && token.Error() != nil {
|
|
log.Printf("Unable to publish, error %s", token.Error())
|
|
}
|
|
log.Println("Successfully published")
|
|
}
|
|
}
|
|
}
|
|
|
|
func StartMqttClient() {
|
|
broker := config.Config.Mqtt.Broker
|
|
if broker == "" {
|
|
log.Fatal("No broker given, set env var MQTT_BROKER")
|
|
}
|
|
|
|
prefix := "UDI"
|
|
uuid := uuid.New()
|
|
clientId := fmt.Sprintf("%s-%s", prefix, uuid)
|
|
|
|
opts := MQTT.NewClientOptions().
|
|
AddBroker(broker).
|
|
SetClientID(clientId).
|
|
SetConnectionLostHandler(onConnectionLost).
|
|
SetOnConnectHandler(onConnect).
|
|
SetReconnectingHandler(onReconnecting).
|
|
SetConnectRetry(true)
|
|
|
|
username := config.Config.Mqtt.Username
|
|
if username != "" {
|
|
opts.SetUsername(username)
|
|
}
|
|
|
|
password := config.Config.Mqtt.Password
|
|
if password != "" {
|
|
opts.SetPassword(password)
|
|
}
|
|
|
|
enableTls := config.Config.Mqtt.TlsEnable
|
|
if enableTls == "true" {
|
|
//log.Println("Enabling TLS connection")
|
|
tlsConfig := &tls.Config {
|
|
InsecureSkipVerify: true,
|
|
}
|
|
opts.SetTLSConfig(tlsConfig)
|
|
}
|
|
|
|
log.Println("Broker connecting")
|
|
mqttClient = MQTT.NewClient(opts)
|
|
if token := mqttClient.Connect(); token.Wait() && token.Error() != nil {
|
|
log.Fatalf("Unable to connect to broker %s, error %s", broker, token.Error())
|
|
}
|
|
//log.Printf("Successfully connected to broker %s", broker)
|
|
|
|
go outputDispatcher(mqttClient)
|
|
|
|
return
|
|
}
|
|
|
|
func StopMqttClient() {
|
|
log.Println("Disconnecting from broker")
|
|
mqttClient.Disconnect(250)
|
|
}
|
|
|
|
func TopicMatchesSubscription(topic, subscription string) bool {
|
|
topicSegments := strings.Split(topic, "/")
|
|
subscriptionSegments := strings.Split(subscription, "/")
|
|
|
|
for i, subSegment := range subscriptionSegments {
|
|
if subSegment == "+" {
|
|
continue
|
|
} else if subSegment == "#" {
|
|
return true
|
|
} else if i < len(topicSegments) && subSegment != topicSegments[i] {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|