11package mqttpubsub
22
33import (
4+ "bytes"
45 "crypto/tls"
56 "crypto/x509"
67 "encoding/json"
78 "fmt"
89 "io/ioutil"
910 "sync"
11+ "text/template"
1012 "time"
1113
1214 "github.com/brocaar/loraserver/api/gw"
1315 "github.com/brocaar/lorawan"
1416 "github.com/eclipse/paho.mqtt.golang"
17+ "github.com/pkg/errors"
1518 log "github.com/sirupsen/logrus"
1619)
1720
@@ -21,15 +24,42 @@ type Backend struct {
2124 txPacketChan chan gw.TXPacketBytes
2225 gateways map [lorawan.EUI64 ]struct {}
2326 mutex sync.RWMutex
27+
28+ UplinkTemplate * template.Template
29+ DownlinkTemplate * template.Template
30+ StatsTemplate * template.Template
31+ AckTemplate * template.Template
2432}
2533
2634// NewBackend creates a new Backend.
27- func NewBackend (server , username , password , cafile , certFile , certKeyFile string ) (* Backend , error ) {
35+ func NewBackend (server , username , password , cafile , certFile , certKeyFile , uplinkTopic , downlinkTopic , statsTopic , ackTopic string ) (* Backend , error ) {
36+ var err error
37+
2838 b := Backend {
2939 txPacketChan : make (chan gw.TXPacketBytes ),
3040 gateways : make (map [lorawan.EUI64 ]struct {}),
3141 }
3242
43+ b .UplinkTemplate , err = template .New ("uplink" ).Parse (uplinkTopic )
44+ if err != nil {
45+ return nil , errors .Wrap (err , "parse uplink template error" )
46+ }
47+
48+ b .DownlinkTemplate , err = template .New ("downlink" ).Parse (downlinkTopic )
49+ if err != nil {
50+ return nil , errors .Wrap (err , "parse downlink template error" )
51+ }
52+
53+ b .StatsTemplate , err = template .New ("stats" ).Parse (statsTopic )
54+ if err != nil {
55+ return nil , errors .Wrap (err , "parse stats template error" )
56+ }
57+
58+ b .AckTemplate , err = template .New ("ack" ).Parse (ackTopic )
59+ if err != nil {
60+ return nil , errors .Wrap (err , "parse ack template error" )
61+ }
62+
3363 opts := mqtt .NewClientOptions ()
3464 opts .AddBroker (server )
3565 opts .SetUsername (username )
@@ -114,9 +144,13 @@ func (b *Backend) SubscribeGatewayTX(mac lorawan.EUI64) error {
114144 defer b .mutex .Unlock ()
115145 b .mutex .Lock ()
116146
117- topic := fmt .Sprintf ("gateway/%s/tx" , mac .String ())
118- log .WithField ("topic" , topic ).Info ("backend: subscribing to topic" )
119- if token := b .conn .Subscribe (topic , 0 , b .txPacketHandler ); token .Wait () && token .Error () != nil {
147+ topic := bytes .NewBuffer (nil )
148+ if err := b .DownlinkTemplate .Execute (topic , struct { MAC lorawan.EUI64 }{mac }); err != nil {
149+ return errors .Wrap (err , "execute uplink template error" )
150+ }
151+
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 {
120154 return token .Error ()
121155 }
122156 b .gateways [mac ] = struct {}{}
@@ -129,9 +163,13 @@ func (b *Backend) UnSubscribeGatewayTX(mac lorawan.EUI64) error {
129163 defer b .mutex .Unlock ()
130164 b .mutex .Lock ()
131165
132- topic := fmt .Sprintf ("gateway/%s/tx" , mac .String ())
133- log .WithField ("topic" , topic ).Info ("backend: unsubscribing from topic" )
134- if token := b .conn .Unsubscribe (topic ); token .Wait () && token .Error () != nil {
166+ topic := bytes .NewBuffer (nil )
167+ if err := b .DownlinkTemplate .Execute (topic , struct { MAC lorawan.EUI64 }{mac }); err != nil {
168+ return errors .Wrap (err , "execute uplink template error" )
169+ }
170+
171+ log .WithField ("topic" , topic .String ()).Info ("backend: unsubscribing from topic" )
172+ if token := b .conn .Unsubscribe (topic .String ()); token .Wait () && token .Error () != nil {
135173 return token .Error ()
136174 }
137175 delete (b .gateways , mac )
@@ -140,29 +178,31 @@ func (b *Backend) UnSubscribeGatewayTX(mac lorawan.EUI64) error {
140178
141179// PublishGatewayRX publishes a RX packet to the MQTT broker.
142180func (b * Backend ) PublishGatewayRX (mac lorawan.EUI64 , rxPacket gw.RXPacketBytes ) error {
143- topic := fmt .Sprintf ("gateway/%s/rx" , mac .String ())
144- return b .publish (topic , rxPacket )
181+ return b .publish (mac , b .UplinkTemplate , rxPacket )
145182}
146183
147184// PublishGatewayStats publishes a GatewayStatsPacket to the MQTT broker.
148185func (b * Backend ) PublishGatewayStats (mac lorawan.EUI64 , stats gw.GatewayStatsPacket ) error {
149- topic := fmt .Sprintf ("gateway/%s/stats" , mac .String ())
150- return b .publish (topic , stats )
186+ return b .publish (mac , b .StatsTemplate , stats )
151187}
152188
153189// PublishGatewayTXAck publishes a TX ack to the MQTT broker.
154190func (b * Backend ) PublishGatewayTXAck (mac lorawan.EUI64 , ack gw.TXAck ) error {
155- topic := fmt .Sprintf ("gateway/%s/ack" , mac .String ())
156- return b .publish (topic , ack )
191+ return b .publish (mac , b .AckTemplate , ack )
157192}
158193
159- func (b * Backend ) publish (topic string , v interface {}) error {
194+ func (b * Backend ) publish (mac lorawan.EUI64 , topicTemplate * template.Template , v interface {}) error {
195+ topic := bytes .NewBuffer (nil )
196+ if err := topicTemplate .Execute (topic , struct { MAC lorawan.EUI64 }{mac }); err != nil {
197+ return errors .Wrap (err , "execute template error" )
198+ }
199+
160200 bytes , err := json .Marshal (v )
161201 if err != nil {
162202 return err
163203 }
164- log .WithField ("topic" , topic ).Info ("backend: publishing packet" )
165- if token := b .conn .Publish (topic , 0 , false , bytes ); token .Wait () && token .Error () != nil {
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 {
166206 return token .Error ()
167207 }
168208 return nil
0 commit comments