Skip to content

Commit 7db3f5d

Browse files
brocaarbrocaar
committed
Expose MQTT client id, qos and clean-session options.
Closes #24.
1 parent 6f5a05e commit 7db3f5d

File tree

9 files changed

+131
-62
lines changed

9 files changed

+131
-62
lines changed

‎cmd/lora-gateway-bridge/cmd/configfile.go

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,30 @@ username="{{ .Backend.MQTT.Username }}"
5757
# Connect with the given password (optional)
5858
password="{{ .Backend.MQTT.Password }}"
5959
60+
# Quality of service level
61+
#
62+
# 0: at most once
63+
# 1: at least once
64+
# 2: exactly once
65+
#
66+
# Note: an increase of this value will decrease the performance.
67+
# For more information: https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels
68+
qos={{ .Backend.MQTT.QOS }}
69+
70+
# Clean session
71+
#
72+
# Set the "clean session" flag in the connect message when this client
73+
# connects to an MQTT broker. By setting this flag you are indicating
74+
# that no messages saved by the broker for this client should be delivered.
75+
clean_session={{ .Backend.MQTT.CleanSession }}
76+
77+
# Client ID
78+
#
79+
# Set the client id to be used by this client when connecting to the MQTT
80+
# broker. A client id must be no longer than 23 characters. When left blank,
81+
# a random id will be generated. This requires clean_session=true.
82+
client_id="{{ .Backend.MQTT.ClientID }}"
83+
6084
# CA certificate file (optional)
6185
#
6286
# Use this when setting up a secure connection (when server uses ssl://...)

‎cmd/lora-gateway-bridge/cmd/root.go

Lines changed: 2 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ func init() {
8484
viper.SetDefault("backend.mqtt.stats_topic_template", "gateway/{{ .MAC }}/stats")
8585
viper.SetDefault("backend.mqtt.ack_topic_template", "gateway/{{ .MAC }}/ack")
8686
viper.SetDefault("backend.mqtt.server", "tcp://127.0.0.1:1883")
87+
viper.SetDefault("backend.mqtt.clean_session", true)
8788

8889
rootCmd.AddCommand(versionCmd)
8990
rootCmd.AddCommand(configCmd)
@@ -108,18 +109,7 @@ func run(cmd *cobra.Command, args []string) error {
108109
var pubsub *mqttpubsub.Backend
109110
for {
110111
var err error
111-
pubsub, err = mqttpubsub.NewBackend(
112-
config.C.Backend.MQTT.Server,
113-
config.C.Backend.MQTT.Username,
114-
config.C.Backend.MQTT.Password,
115-
config.C.Backend.MQTT.CACert,
116-
config.C.Backend.MQTT.TLSCert,
117-
config.C.Backend.MQTT.TLSKey,
118-
config.C.Backend.MQTT.UplinkTopicTemplate,
119-
config.C.Backend.MQTT.DownlinkTopicTemplate,
120-
config.C.Backend.MQTT.StatsTopicTemplate,
121-
config.C.Backend.MQTT.AckTopicTemplate,
122-
)
112+
pubsub, err = mqttpubsub.NewBackend(config.C.Backend.MQTT)
123113
if err == nil {
124114
break
125115
}

‎docs/config.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,4 +32,4 @@ googleAnalytics = "UA-3512995-9"
3232
weight = 4
3333

3434
[params]
35-
version = "2.3.1"
35+
version = "2.3.2"

‎docs/content/install/config.md

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,7 @@ skip_crc_check = false
9797

9898
# Configuration for the MQTT backend.
9999
[backend.mqtt]
100-
# MQTT topic templates for the different MQTT topic.
100+
# MQTT topic templates for the different MQTT topics.
101101
#
102102
# The meaning of these topics are documented at:
103103
# https://docs.loraserver.io/lora-gateway-bridge/use/data/
@@ -120,6 +120,30 @@ username=""
120120
# Connect with the given password (optional)
121121
password=""
122122

123+
# Quality of service level
124+
#
125+
# 0: at most once
126+
# 1: at least once
127+
# 2: exactly once
128+
#
129+
# Note: an increase of this value will decrease the performance.
130+
# For more information: https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels
131+
qos=0
132+
133+
# Clean session
134+
#
135+
# Set the "clean session" flag in the connect message when this client
136+
# connects to an MQTT broker. By setting this flag you are indicating
137+
# that no messages saved by the broker for this client should be delivered.
138+
clean_session=true
139+
140+
# Client ID
141+
#
142+
# Set the client id to be used by this client when connecting to the MQTT
143+
# broker. A client id must be no longer than 23 characters. When left blank,
144+
# a random id will be generated. This requires clean_session=true.
145+
client_id=""
146+
123147
# CA certificate file (optional)
124148
#
125149
# Use this when setting up a secure connection (when server uses ssl://...)

‎docs/content/overview/changelog.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,19 @@ menu:
88

99
## Changelog
1010

11+
### 2.3.2
12+
13+
**Improvements:**
14+
15+
* Expose the following MQTT options for the MQTT gateway backend:
16+
* QoS (quality of service)
17+
* Client ID
18+
* Clean session on connect
19+
20+
**Bugfixes:**
21+
22+
* Use topic from configuration file on re-connect (this was still hardcoded).
23+
1124
### 2.3.1
1225

1326
**Improvements:**

‎internal/backend/mqttpubsub/backend.go

Lines changed: 53 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"crypto/tls"
66
"crypto/x509"
77
"encoding/json"
8-
"fmt"
98
"io/ioutil"
109
"sync"
1110
"text/template"
@@ -18,12 +17,30 @@ import (
1817
log "github.com/sirupsen/logrus"
1918
)
2019

20+
// BackendConfig holds the MQTT pub-sub backend configuration.
21+
type BackendConfig struct {
22+
Server string
23+
Username string
24+
Password string
25+
QOS uint8 `mapstructure:"qos"`
26+
CleanSession bool `mapstructure:"clean_session"`
27+
ClientID string `mapstructure:"client_id"`
28+
CACert string `mapstructure:"ca_cert"`
29+
TLSCert string `mapstructure:"tls_cert"`
30+
TLSKey string `mapstructure:"tls_key"`
31+
UplinkTopicTemplate string `mapstructure:"uplink_topic_template"`
32+
DownlinkTopicTemplate string `mapstructure:"downlink_topic_template"`
33+
StatsTopicTemplate string `mapstructure:"stats_topic_template"`
34+
AckTopicTemplate string `mapstructure:"ack_topic_template"`
35+
}
36+
2137
// Backend implements a MQTT pub-sub backend.
2238
type Backend struct {
2339
conn mqtt.Client
2440
txPacketChan chan gw.TXPacketBytes
2541
gateways map[lorawan.EUI64]struct{}
2642
mutex sync.RWMutex
43+
config BackendConfig
2744

2845
UplinkTemplate *template.Template
2946
DownlinkTemplate *template.Template
@@ -32,54 +49,57 @@ type Backend struct {
3249
}
3350

3451
// NewBackend creates a new Backend.
35-
func NewBackend(server, username, password, cafile, certFile, certKeyFile, uplinkTopic, downlinkTopic, statsTopic, ackTopic string) (*Backend, error) {
52+
func NewBackend(c BackendConfig) (*Backend, error) {
3653
var err error
3754

3855
b := Backend{
3956
txPacketChan: make(chan gw.TXPacketBytes),
4057
gateways: make(map[lorawan.EUI64]struct{}),
58+
config: c,
4159
}
4260

43-
b.UplinkTemplate, err = template.New("uplink").Parse(uplinkTopic)
61+
b.UplinkTemplate, err = template.New("uplink").Parse(b.config.UplinkTopicTemplate)
4462
if err != nil {
4563
return nil, errors.Wrap(err, "parse uplink template error")
4664
}
4765

48-
b.DownlinkTemplate, err = template.New("downlink").Parse(downlinkTopic)
66+
b.DownlinkTemplate, err = template.New("downlink").Parse(b.config.DownlinkTopicTemplate)
4967
if err != nil {
5068
return nil, errors.Wrap(err, "parse downlink template error")
5169
}
5270

53-
b.StatsTemplate, err = template.New("stats").Parse(statsTopic)
71+
b.StatsTemplate, err = template.New("stats").Parse(b.config.StatsTopicTemplate)
5472
if err != nil {
5573
return nil, errors.Wrap(err, "parse stats template error")
5674
}
5775

58-
b.AckTemplate, err = template.New("ack").Parse(ackTopic)
76+
b.AckTemplate, err = template.New("ack").Parse(b.config.AckTopicTemplate)
5977
if err != nil {
6078
return nil, errors.Wrap(err, "parse ack template error")
6179
}
6280

6381
opts := mqtt.NewClientOptions()
64-
opts.AddBroker(server)
65-
opts.SetUsername(username)
66-
opts.SetPassword(password)
82+
opts.AddBroker(b.config.Server)
83+
opts.SetUsername(b.config.Username)
84+
opts.SetPassword(b.config.Password)
85+
opts.SetCleanSession(b.config.CleanSession)
86+
opts.SetClientID(b.config.ClientID)
6787
opts.SetOnConnectHandler(b.onConnected)
6888
opts.SetConnectionLostHandler(b.onConnectionLost)
6989

70-
tlsconfig, err := NewTLSConfig(cafile, certFile, certKeyFile)
90+
tlsconfig, err := NewTLSConfig(b.config.CACert, b.config.TLSCert, b.config.TLSKey)
7191
if err != nil {
7292
log.WithError(err).WithFields(log.Fields{
73-
"ca_cert": cafile,
74-
"tls_cert": certFile,
75-
"tls_key": certKeyFile,
93+
"ca_cert": b.config.CACert,
94+
"tls_cert": b.config.TLSCert,
95+
"tls_key": b.config.TLSKey,
7696
}).Fatal("error loading mqtt certificate files")
7797
}
7898
if tlsconfig != nil {
7999
opts.SetTLSConfig(tlsconfig)
80100
}
81101

82-
log.WithField("server", server).Info("backend: connecting to mqtt broker")
102+
log.WithField("server", b.config.Server).Info("backend: connecting to mqtt broker")
83103
b.conn = mqtt.NewClient(opts)
84104
if token := b.conn.Connect(); token.Wait() && token.Error() != nil {
85105
return nil, token.Error()
@@ -146,11 +166,14 @@ func (b *Backend) SubscribeGatewayTX(mac lorawan.EUI64) error {
146166

147167
topic := bytes.NewBuffer(nil)
148168
if err := b.DownlinkTemplate.Execute(topic, struct{ MAC lorawan.EUI64 }{mac}); err != nil {
149-
return errors.Wrap(err, "execute uplink template error")
169+
return errors.Wrap(err, "execute downlink template error")
150170
}
151171

152-
log.WithField("topic", topic.String()).Info("backend: subscribing to topic")
153-
if token := b.conn.Subscribe(topic.String(), 0, b.txPacketHandler); token.Wait() && token.Error() != nil {
172+
log.WithFields(log.Fields{
173+
"topic": topic.String(),
174+
"qos": b.config.QOS,
175+
}).Info("backend: subscribing to topic")
176+
if token := b.conn.Subscribe(topic.String(), b.config.QOS, b.txPacketHandler); token.Wait() && token.Error() != nil {
154177
return token.Error()
155178
}
156179
b.gateways[mac] = struct{}{}
@@ -165,7 +188,7 @@ func (b *Backend) UnSubscribeGatewayTX(mac lorawan.EUI64) error {
165188

166189
topic := bytes.NewBuffer(nil)
167190
if err := b.DownlinkTemplate.Execute(topic, struct{ MAC lorawan.EUI64 }{mac}); err != nil {
168-
return errors.Wrap(err, "execute uplink template error")
191+
return errors.Wrap(err, "execute downlink template error")
169192
}
170193

171194
log.WithField("topic", topic.String()).Info("backend: unsubscribing from topic")
@@ -201,8 +224,11 @@ func (b *Backend) publish(mac lorawan.EUI64, topicTemplate *template.Template, v
201224
if err != nil {
202225
return err
203226
}
204-
log.WithField("topic", topic.String()).Info("backend: publishing packet")
205-
if token := b.conn.Publish(topic.String(), 0, false, bytes); token.Wait() && token.Error() != nil {
227+
log.WithFields(log.Fields{
228+
"topic": topic.String(),
229+
"qos": b.config.QOS,
230+
}).Info("backend: publishing packet")
231+
if token := b.conn.Publish(topic.String(), b.config.QOS, false, bytes); token.Wait() && token.Error() != nil {
206232
return token.Error()
207233
}
208234
return nil
@@ -228,7 +254,13 @@ func (b *Backend) onConnected(c mqtt.Client) {
228254
log.WithField("topic_count", len(b.gateways)).Info("backend: re-registering to gateway topics")
229255
topics := make(map[string]byte)
230256
for k := range b.gateways {
231-
topics[fmt.Sprintf("gateway/%s/tx", k)] = 0
257+
topic := bytes.NewBuffer(nil)
258+
if err := b.DownlinkTemplate.Execute(topic, struct{ MAC lorawan.EUI64 }{k}); err != nil {
259+
log.WithError(err).Error("backend: execute downlink template error")
260+
time.Sleep(time.Second)
261+
continue
262+
}
263+
topics[topic.String()] = b.config.QOS
232264
}
233265
if token := b.conn.SubscribeMultiple(topics, b.txPacketHandler); token.Wait() && token.Error() != nil {
234266
log.WithField("topic_count", len(topics)).Errorf("backend: subscribe multiple failed: %s", token.Error())

‎internal/backend/mqttpubsub/backend_test.go

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"testing"
66
"time"
77

8-
"github.com/brocaar/lora-gateway-bridge/internal/config"
98
"github.com/brocaar/loraserver/api/gw"
109
"github.com/eclipse/paho.mqtt.golang"
1110
. "github.com/smartystreets/goconvey/convey"
@@ -24,14 +23,16 @@ func TestBackend(t *testing.T) {
2423

2524
Convey("Given a new Backend", func() {
2625
backend, err := NewBackend(
27-
conf.Server,
28-
conf.Username,
29-
conf.Password,
30-
"", "", "",
31-
config.C.Backend.MQTT.UplinkTopicTemplate,
32-
config.C.Backend.MQTT.DownlinkTopicTemplate,
33-
config.C.Backend.MQTT.StatsTopicTemplate,
34-
config.C.Backend.MQTT.AckTopicTemplate,
26+
BackendConfig{
27+
Server: conf.Server,
28+
Username: conf.Username,
29+
Password: conf.Password,
30+
CleanSession: true,
31+
UplinkTopicTemplate: "gateway/{{ .MAC }}/rx",
32+
DownlinkTopicTemplate: "gateway/{{ .MAC }}/tx",
33+
StatsTopicTemplate: "gateway/{{ .MAC }}/stats",
34+
AckTopicTemplate: "gateway/{{ .MAC }}/ack",
35+
},
3536
)
3637
So(err, ShouldBeNil)
3738
defer backend.Close()

‎internal/backend/mqttpubsub/base_test.go

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package mqttpubsub
33
import (
44
"os"
55

6-
"github.com/brocaar/lora-gateway-bridge/internal/config"
76
log "github.com/sirupsen/logrus"
87
)
98

@@ -18,11 +17,6 @@ type conf struct {
1817
}
1918

2019
func getConfig() *conf {
21-
config.C.Backend.MQTT.DownlinkTopicTemplate = "gateway/{{ .MAC }}/tx"
22-
config.C.Backend.MQTT.UplinkTopicTemplate = "gateway/{{ .MAC }}/rx"
23-
config.C.Backend.MQTT.StatsTopicTemplate = "gateway/{{ .MAC }}/stats"
24-
config.C.Backend.MQTT.AckTopicTemplate = "gateway/{{ .MAC }}/ack"
25-
2620
c := &conf{
2721
Server: "tcp://127.0.0.1:1883",
2822
}

‎internal/config/config.go

Lines changed: 3 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package config
22

3+
import "github.com/brocaar/lora-gateway-bridge/internal/backend/mqttpubsub"
4+
35
// Config defines the configuration structure.
46
type Config struct {
57
General struct {
@@ -12,18 +14,7 @@ type Config struct {
1214
} `mapstructure:"packet_forwarder"`
1315

1416
Backend struct {
15-
MQTT struct {
16-
Server string
17-
Username string
18-
Password string
19-
CACert string `mapstructure:"ca_cert"`
20-
TLSCert string `mapstructure:"tls_cert"`
21-
TLSKey string `mapstructure:"tls_key"`
22-
UplinkTopicTemplate string `mapstructure:"uplink_topic_template"`
23-
DownlinkTopicTemplate string `mapstructure:"downlink_topic_template"`
24-
StatsTopicTemplate string `mapstructure:"stats_topic_template"`
25-
AckTopicTemplate string `mapstructure:"ack_topic_template"`
26-
}
17+
MQTT mqttpubsub.BackendConfig
2718
}
2819
}
2920

0 commit comments

Comments
 (0)