Implement unique IDs (UUID) for events and commands.

Author: brocaar

go.mod

@@ -3,21 +3,20 @@ module github.com/brocaar/lora-gateway-bridge
 go 1.12
 
 require (
-	github.com/brocaar/loraserver v0.0.0-20190729122155-2f0bb9c308bc
+	github.com/brocaar/loraserver v0.0.0-20190909075954-156cedb76807
 	github.com/brocaar/lorawan v0.0.0-20190814113539-8eb2a8d6da09
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.2.0
+	github.com/gofrs/uuid v3.2.0+incompatible
 	github.com/golang/protobuf v1.3.2
 	github.com/goreleaser/goreleaser v0.106.0
 	github.com/gorilla/websocket v1.4.0
 	github.com/pkg/errors v0.8.1
-	github.com/prometheus/client_golang v1.0.0
-	github.com/prometheus/common v0.6.0 // indirect
-	github.com/prometheus/procfs v0.0.3 // indirect
+	github.com/prometheus/client_golang v1.1.0
 	github.com/sirupsen/logrus v1.4.2
-	github.com/spf13/cobra v0.0.3
-	github.com/spf13/viper v1.3.2
-	github.com/stretchr/testify v1.3.0
+	github.com/spf13/cobra v0.0.5
+	github.com/spf13/viper v1.4.0
+	github.com/stretchr/testify v1.4.0
 	golang.org/x/lint v0.0.0-20190409202823-959b441ac422
 	golang.org/x/tools v0.0.0-20190709211700-7b25e351ac0e // indirect
 )

go.sum

@@ -1,8 +1,10 @@
 package basicstation
 
 import (
+	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -12,6 +14,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gofrs/uuid"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/gorilla/websocket"
 	"github.com/pkg/errors"
@@ -55,6 +58,11 @@ type Backend struct {
 	joinEUIs     [][2]lorawan.EUI64
 	frequencyMin uint32
 	frequencyMax uint32
+
+	// diidMap stores the mapping of diid to UUIDs. This should take ~ 1MB of
+	// memory. Optionaly this could be optimized by letting keys expire after
+	// a given time.
+	diidMap map[uint16][]byte
 }
 
 // NewBackend creates a new Backend.
@@ -79,6 +87,8 @@ func NewBackend(conf config.Config) (*Backend, error) {
 		region:       band.Name(conf.Backend.BasicStation.Region),
 		frequencyMin: conf.Backend.BasicStation.FrequencyMin,
 		frequencyMax: conf.Backend.BasicStation.FrequencyMax,
+
+		diidMap: make(map[uint16][]byte),
 	}
 
 	for _, n := range conf.Filters.NetIDs {
@@ -191,20 +201,41 @@ func (b *Backend) GetDisconnectChan() chan lorawan.EUI64 {
 }
 
 func (b *Backend) SendDownlinkFrame(df gw.DownlinkFrame) error {
+	b.Lock()
+	defer b.Unlock()
+
+	// for backwards compatibility
+	if df.Token == 0 {
+		tokenB := make([]byte, 2)
+		if _, err := rand.Read(tokenB); err != nil {
+			return errors.Wrap(err, "read random bytes error")
+		}
+
+		df.Token = uint32(binary.BigEndian.Uint16(tokenB))
+	}
+
 	pl, err := structs.DownlinkFrameFromProto(b.band, df)
 	if err != nil {
 		return errors.Wrap(err, "downlink frame from proto error")
 	}
 
 	var gatewayID lorawan.EUI64
-	copy(gatewayID[:], df.TxInfo.GatewayId)
+	var downID uuid.UUID
+	copy(gatewayID[:], df.GetTxInfo().GetGatewayId())
+	copy(downID[:], df.GetDownlinkId())
+
+	// store token to UUID mapping
+	b.diidMap[uint16(df.Token)] = df.GetDownlinkId()
 
 	websocketSendCounter("dnmsg").Inc()
 	if err := b.sendToGateway(gatewayID, pl); err != nil {
 		return errors.Wrap(err, "send to gateway error")
 	}
 
-	log.WithField("gateway_id", gatewayID).Info("backend/basicstation: downlink-frame message sent to gateway")
+	log.WithFields(log.Fields{
+		"gateway_id":  gatewayID,
+		"downlink_id": downID,
+	}).Info("backend/basicstation: downlink-frame message sent to gateway")
 
 	return nil
 }
@@ -216,7 +247,7 @@ func (b *Backend) ApplyConfiguration(gwConfig gw.GatewayConfiguration) error {
 	}
 
 	var gatewayID lorawan.EUI64
-	copy(gatewayID[:], gwConfig.GatewayId)
+	copy(gatewayID[:], gwConfig.GetGatewayId())
 
 	websocketSendCounter("router_config").Inc()
 	if err := b.sendToGateway(gatewayID, rc); err != nil {
@@ -292,14 +323,13 @@ func (b *Backend) handleGateway(r *http.Request, c *websocket.Conn) {
 		var cn lorawan.EUI64
 		if err := cn.UnmarshalText([]byte(r.TLS.PeerCertificates[0].Subject.CommonName)); err != nil || cn != gatewayID {
 			log.WithFields(log.Fields{
-				"gateway_id": gatewayID,
+				"gateway_id":  gatewayID,
 				"common_name": r.TLS.PeerCertificates[0].Subject.CommonName,
 			}).Error("backend/basicstation: CommonName verification failed")
 			return
 		}
 	}
 
-
 	// make sure we're not overwriting an existing connection
 	_, err := b.gateways.get(gatewayID)
 	if err == nil {
@@ -459,10 +489,6 @@ func (b *Backend) handleVersion(gatewayID lorawan.EUI64, pl structs.Version) {
 }
 
 func (b *Backend) handleJoinRequest(gatewayID lorawan.EUI64, v structs.JoinRequest) {
-	log.WithFields(log.Fields{
-		"gateway_id": gatewayID,
-	}).Info("backend/basicstation: join-request received")
-
 	uplinkFrame, err := structs.JoinRequestToProto(b.band, gatewayID, v)
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
@@ -471,14 +497,25 @@ func (b *Backend) handleJoinRequest(gatewayID lorawan.EUI64, v structs.JoinReque
 		return
 	}
 
-	b.uplinkFrameChan <- uplinkFrame
-}
+	// set uplink id
+	uplinkID, err := uuid.NewV4()
+	if err != nil {
+		log.WithError(err).WithFields(log.Fields{
+			"gateway_id": gatewayID,
+		}).Error("backend/basicstation: get random uplink id error")
+		return
+	}
+	uplinkFrame.RxInfo.UplinkId = uplinkID[:]
 
-func (b *Backend) handleProprietaryDataFrame(gatewayID lorawan.EUI64, v structs.UplinkProprietaryFrame) {
 	log.WithFields(log.Fields{
 		"gateway_id": gatewayID,
-	}).Info("backend/basicstation: proprietary uplink frame received")
+		"uplink_id":  uplinkID,
+	}).Info("backend/basicstation: join-request received")
+
+	b.uplinkFrameChan <- uplinkFrame
+}
 
+func (b *Backend) handleProprietaryDataFrame(gatewayID lorawan.EUI64, v structs.UplinkProprietaryFrame) {
 	uplinkFrame, err := structs.UplinkProprietaryFrameToProto(b.band, gatewayID, v)
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
@@ -487,13 +524,27 @@ func (b *Backend) handleProprietaryDataFrame(gatewayID lorawan.EUI64, v structs.
 		return
 	}
 
+	// set uplink id
+	uplinkID, err := uuid.NewV4()
+	if err != nil {
+		log.WithError(err).WithFields(log.Fields{
+			"gateway_id": gatewayID,
+		}).Error("backend/basicstation: get random uplink id error")
+		return
+	}
+	uplinkFrame.RxInfo.UplinkId = uplinkID[:]
+
+	log.WithFields(log.Fields{
+		"gateway_id": gatewayID,
+		"uplink_id":  uplinkID,
+	}).Info("backend/basicstation: proprietary uplink frame received")
+
 	b.uplinkFrameChan <- uplinkFrame
 }
 
 func (b *Backend) handleDownlinkTransmittedMessage(gatewayID lorawan.EUI64, v structs.DownlinkTransmitted) {
-	log.WithFields(log.Fields{
-		"gateway_id": gatewayID,
-	}).Info("backend/basicstation: downlink transmitted message received")
+	b.RLock()
+	defer b.RUnlock()
 
 	txack, err := structs.DownlinkTransmittedToProto(gatewayID, v)
 	if err != nil {
@@ -502,15 +553,20 @@ func (b *Backend) handleDownlinkTransmittedMessage(gatewayID lorawan.EUI64, v st
 		}).Error("backend/basicstation: error converting downlink transmitted to protobuf message")
 		return
 	}
+	txack.DownlinkId = b.diidMap[uint16(v.DIID)]
+
+	var downID uuid.UUID
+	copy(downID[:], txack.GetDownlinkId())
+
+	log.WithFields(log.Fields{
+		"gateway_id":  gatewayID,
+		"downlink_id": downID,
+	}).Info("backend/basicstation: downlink transmitted message received")
 
 	b.downlinkTXAckChan <- txack
 }
 
 func (b *Backend) handleUplinkDataFrame(gatewayID lorawan.EUI64, v structs.UplinkDataFrame) {
-	log.WithFields(log.Fields{
-		"gateway_id": gatewayID,
-	}).Info("backend/basicstation: uplink frame received")
-
 	uplinkFrame, err := structs.UplinkDataFrameToProto(b.band, gatewayID, v)
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
@@ -519,6 +575,21 @@ func (b *Backend) handleUplinkDataFrame(gatewayID lorawan.EUI64, v structs.Uplin
 		return
 	}
 
+	// set uplink id
+	uplinkID, err := uuid.NewV4()
+	if err != nil {
+		log.WithError(err).WithFields(log.Fields{
+			"gateway_id": gatewayID,
+		}).Error("backend/basicstation: get random uplink id error")
+		return
+	}
+	uplinkFrame.RxInfo.UplinkId = uplinkID[:]
+
+	log.WithFields(log.Fields{
+		"gateway_id": gatewayID,
+		"uplink_id":  uplinkID,
+	}).Info("backend/basicstation: uplink frame received")
+
 	b.uplinkFrameChan <- uplinkFrame
 }
 

internal/backend/basicstation/backend.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gofrs/uuid"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/gorilla/websocket"
 	log "github.com/sirupsen/logrus"
@@ -136,6 +137,10 @@ func (ts *BackendTestSuite) TestUplinkDataFrame() {
 	assert.NoError(ts.wsClient.WriteJSON(upf))
 
 	uplinkFrame := <-ts.backend.GetUplinkFrameChan()
+
+	assert.Len(uplinkFrame.RxInfo.UplinkId, 16)
+	uplinkFrame.RxInfo.UplinkId = nil
+
 	assert.Equal(gw.UplinkFrame{
 		PhyPayload: []byte{0x40, 0xf6, 0xff, 0xff, 0x0ff, 0x80, 0x90, 0x01, 0x01, 0x02, 0xec, 0xff, 0xff, 0xff},
 		TxInfo: &gw.UplinkTXInfo{
@@ -184,6 +189,10 @@ func (ts *BackendTestSuite) TestJoinRequest() {
 	assert.NoError(ts.wsClient.WriteJSON(jr))
 
 	uplinkFrame := <-ts.backend.GetUplinkFrameChan()
+
+	assert.Len(uplinkFrame.RxInfo.UplinkId, 16)
+	uplinkFrame.RxInfo.UplinkId = nil
+
 	assert.Equal(gw.UplinkFrame{
 		PhyPayload: []byte{0x00, 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x02, 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x03, 0x14, 0x00, 0xf6, 0xff, 0xff, 0xff},
 		TxInfo: &gw.UplinkTXInfo{
@@ -227,6 +236,10 @@ func (ts *BackendTestSuite) TestProprietaryDataFrame() {
 	assert.NoError(ts.wsClient.WriteJSON(propf))
 
 	uplinkFrame := <-ts.backend.GetUplinkFrameChan()
+
+	assert.Len(uplinkFrame.RxInfo.UplinkId, 16)
+	uplinkFrame.RxInfo.UplinkId = nil
+
 	assert.Equal(gw.UplinkFrame{
 		PhyPayload: []byte{0x01, 0x02, 0x03, 0x04},
 		TxInfo: &gw.UplinkTXInfo{
@@ -251,6 +264,10 @@ func (ts *BackendTestSuite) TestProprietaryDataFrame() {
 
 func (ts *BackendTestSuite) TestDownlinkTransmitted() {
 	assert := require.New(ts.T())
+	id, err := uuid.NewV4()
+	assert.NoError(err)
+
+	ts.backend.diidMap[12345] = id[:]
 
 	dtx := structs.DownlinkTransmitted{
 		MessageType: structs.DownlinkTransmittedMessage,
@@ -260,9 +277,11 @@ func (ts *BackendTestSuite) TestDownlinkTransmitted() {
 	assert.NoError(ts.wsClient.WriteJSON(dtx))
 
 	txAck := <-ts.backend.GetDownlinkTXAckChan()
+
 	assert.Equal(gw.DownlinkTXAck{
-		GatewayId: []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08},
-		Token:     12345,
+		GatewayId:  []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08},
+		Token:      12345,
+		DownlinkId: id[:],
 	}, txAck)
 }
 
@@ -359,8 +378,10 @@ func (ts *BackendTestSuite) TestApplyConfiguration() {
 
 func (ts *BackendTestSuite) TestSendDownlinkFrame() {
 	assert := require.New(ts.T())
+	id, err := uuid.NewV4()
+	assert.NoError(err)
 
-	err := ts.backend.SendDownlinkFrame(gw.DownlinkFrame{
+	err = ts.backend.SendDownlinkFrame(gw.DownlinkFrame{
 		PhyPayload: []byte{1, 2, 3, 4},
 		TxInfo: &gw.DownlinkTXInfo{
 			GatewayId:  []byte{1, 2, 3, 4, 5, 6, 7, 8},
@@ -383,10 +404,13 @@ func (ts *BackendTestSuite) TestSendDownlinkFrame() {
 			},
 			Context: []byte{0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 4},
 		},
-		Token: 1234,
+		Token:      1234,
+		DownlinkId: id[:],
 	})
 	assert.NoError(err)
 
+	assert.Equal(id[:], ts.backend.diidMap[1234])
+
 	var df structs.DownlinkFrame
 	assert.NoError(ts.wsClient.ReadJSON(&df))