Implement raw events and commands for BasicStation backend.

Author: brocaar

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/gopherjs/gopherjs v0.0.0-20190430165422-3e4dfb77656c // indirect
 	github.com/goreleaser/goreleaser v0.106.0
 	github.com/goreleaser/nfpm v0.11.0
-	github.com/gorilla/websocket v1.4.0
+	github.com/gorilla/websocket v1.4.1
 	github.com/jacobsa/crypto v0.0.0-20190317225127-9f44e2d11115 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
 	github.com/pkg/errors v0.8.1

go.sum

@@ -89,6 +89,8 @@ github.com/goreleaser/nfpm v0.11.0 h1:YJ3wyfTJbqHrE3Ym0b4odQYzGuLdemx09wPLafeZAK
 github.com/goreleaser/nfpm v0.11.0/go.mod h1:F2yzin6cBAL9gb+mSiReuXdsfTrOQwDMsuSpULof+y4=
 github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
+github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
+github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=

internal/backend/backend.go

@@ -5,10 +5,10 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/basicstation"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/semtechudp"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
-	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/lorawan"
 )
 
@@ -53,6 +53,9 @@ type Backend interface {
 	// GetUplinkFrameChan returns the channel for received uplinks.
 	GetUplinkFrameChan() chan gw.UplinkFrame
 
+	// GetRawPacketForwarderEventChan returns the raw packet-forwarder command channel.
+	GetRawPacketForwarderEventChan() chan gw.RawPacketForwarderEvent
+
 	// GetConnectChan returns the channel for received gateway connections.
 	GetConnectChan() chan lorawan.EUI64
 
@@ -64,4 +67,7 @@ type Backend interface {
 
 	// ApplyConfiguration applies the given configuration to the gateway.
 	ApplyConfiguration(gw.GatewayConfiguration) error
+
+	// RawPacketForwarderCommand sends the given raw command to the packet-forwarder.
+	RawPacketForwarderCommand(gw.RawPacketForwarderCommand) error
 }

internal/backend/basicstation/backend.go

@@ -4,6 +4,7 @@ import (
 	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/base64"
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
@@ -20,9 +21,9 @@ import (
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/basicstation/structs"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
-	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/lorawan"
 	"github.com/brocaar/lorawan/band"
 )
@@ -48,9 +49,10 @@ type Backend struct {
 
 	gateways gateways
 
-	downlinkTXAckChan chan gw.DownlinkTXAck
-	uplinkFrameChan   chan gw.UplinkFrame
-	gatewayStatsChan  chan gw.GatewayStats
+	downlinkTXAckChan           chan gw.DownlinkTXAck
+	uplinkFrameChan             chan gw.UplinkFrame
+	gatewayStatsChan            chan gw.GatewayStats
+	rawPacketForwarderEventChan chan gw.RawPacketForwarderEvent
 
 	band         band.Band
 	region       band.Name
@@ -77,9 +79,10 @@ func NewBackend(conf config.Config) (*Backend, error) {
 			disconnectChan: make(chan lorawan.EUI64),
 		},
 
-		downlinkTXAckChan: make(chan gw.DownlinkTXAck),
-		uplinkFrameChan:   make(chan gw.UplinkFrame),
-		gatewayStatsChan:  make(chan gw.GatewayStats),
+		downlinkTXAckChan:           make(chan gw.DownlinkTXAck),
+		uplinkFrameChan:             make(chan gw.UplinkFrame),
+		gatewayStatsChan:            make(chan gw.GatewayStats),
+		rawPacketForwarderEventChan: make(chan gw.RawPacketForwarderEvent),
 
 		pingInterval: conf.Backend.BasicStation.PingInterval,
 		readTimeout:  conf.Backend.BasicStation.ReadTimeout,
@@ -210,6 +213,11 @@ func (b *Backend) GetDisconnectChan() chan lorawan.EUI64 {
 	return b.gateways.disconnectChan
 }
 
+// GetRawPacketForwarderEventChan returns the raw packet-forwarder command channel.
+func (b *Backend) GetRawPacketForwarderEventChan() chan gw.RawPacketForwarderEvent {
+	return b.rawPacketForwarderEventChan
+}
+
 func (b *Backend) SendDownlinkFrame(df gw.DownlinkFrame) error {
 	b.Lock()
 	defer b.Unlock()
@@ -269,6 +277,36 @@ func (b *Backend) ApplyConfiguration(gwConfig gw.GatewayConfiguration) error {
 	return nil
 }
 
+// RawPacketForwarderCommand sends the given raw command to the packet-forwarder.
+func (b *Backend) RawPacketForwarderCommand(pl gw.RawPacketForwarderCommand) error {
+	var gatewayID lorawan.EUI64
+	var rawID uuid.UUID
+
+	copy(gatewayID[:], pl.GatewayId)
+	copy(rawID[:], pl.RawId)
+
+	if len(pl.Payload) == 0 {
+		return errors.New("raw packet-forwarder command payload is empty")
+	}
+
+	mt := websocket.BinaryMessage
+	if strings.HasPrefix(string(pl.Payload), "{") {
+		mt = websocket.TextMessage
+	}
+
+	websocketSendCounter("raw").Inc()
+	if err := b.sendRawToGateway(gatewayID, mt, pl.Payload); err != nil {
+		return errors.Wrap(err, "send raw packet-forwarder command to gateway error")
+	}
+
+	log.WithFields(log.Fields{
+		"gateway_id": gatewayID,
+		"raw_id":     rawID,
+	}).Info("backend/basicstation: raw packet-forwarder command sent to gateway")
+
+	return nil
+}
+
 // Close closes the backend.
 func (b *Backend) Close() error {
 	b.isClosed = true
@@ -302,8 +340,14 @@ func (b *Backend) handleRouterInfo(r *http.Request, c *websocket.Conn) {
 		}
 	}
 
+	bb, err := json.Marshal(resp)
+	if err != nil {
+		log.WithError(err).Error("backend/basicstation: marshal json error")
+		return
+	}
+
 	c.SetWriteDeadline(time.Now().Add(b.writeTimeout))
-	if err := c.WriteJSON(resp); err != nil {
+	if err := c.WriteMessage(websocket.TextMessage, bb); err != nil {
 		log.WithError(err).Error("backend/basicstation: websocket send message error")
 		return
 	}
@@ -367,7 +411,7 @@ func (b *Backend) handleGateway(r *http.Request, c *websocket.Conn) {
 
 	// receive data
 	for {
-		_, msg, err := c.ReadMessage()
+		mt, msg, err := c.ReadMessage()
 		if err != nil {
 			if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
 				log.WithField("gateway_id", gatewayID).WithError(err).Error("backend/basicstation: read message error")
@@ -378,6 +422,16 @@ func (b *Backend) handleGateway(r *http.Request, c *websocket.Conn) {
 		// reset the read deadline as the Basic Station doesn't respond to PONG messages (yet)
 		c.SetReadDeadline(time.Now().Add(b.readTimeout))
 
+		if mt == websocket.BinaryMessage {
+			log.WithFields(log.Fields{
+				"gateway_id":     gatewayID,
+				"message_base64": base64.StdEncoding.EncodeToString(msg),
+			}).Debug("backend/basicstation: binary message received")
+
+			b.handleRawPacketForwarderEvent(gatewayID, msg)
+			continue
+		}
+
 		log.WithFields(log.Fields{
 			"gateway_id": gatewayID,
 			"message":    string(msg),
@@ -458,11 +512,7 @@ func (b *Backend) handleGateway(r *http.Request, c *websocket.Conn) {
 			}
 			b.handleDownlinkTransmittedMessage(gatewayID, pl)
 		default:
-			log.WithFields(log.Fields{
-				"message_type": msgType,
-				"gateway_id":   gatewayID,
-				"payload":      string(msg),
-			}).Warning("backend/basicstation: unexpected message-type")
+			b.handleRawPacketForwarderEvent(gatewayID, msg)
 		}
 	}
 }
@@ -616,14 +666,56 @@ func (b *Backend) handleUplinkDataFrame(gatewayID lorawan.EUI64, v structs.Uplin
 	b.uplinkFrameChan <- uplinkFrame
 }
 
+func (b *Backend) handleRawPacketForwarderEvent(gatewayID lorawan.EUI64, pl []byte) {
+	rawID, err := uuid.NewV4()
+	if err != nil {
+		log.WithError(err).WithFields(log.Fields{
+			"gateway_id": gatewayID,
+		}).Error("backend/basicstation: get random raw id error")
+		return
+	}
+
+	rawEvent := gw.RawPacketForwarderEvent{
+		GatewayId: gatewayID[:],
+		RawId:     rawID[:],
+		Payload:   pl,
+	}
+
+	log.WithFields(log.Fields{
+		"gateway_id": gatewayID,
+		"raw_id":     rawID,
+	}).Info("backend/basicstation: raw packet-forwarder event received")
+
+	b.rawPacketForwarderEventChan <- rawEvent
+}
+
 func (b *Backend) sendToGateway(gatewayID lorawan.EUI64, v interface{}) error {
 	gw, err := b.gateways.get(gatewayID)
 	if err != nil {
 		return errors.Wrap(err, "get gateway error")
 	}
 
+	bb, err := json.Marshal(v)
+	if err != nil {
+		return errors.Wrap(err, "marshal json error")
+	}
+
+	gw.conn.SetWriteDeadline(time.Now().Add(b.writeTimeout))
+	if err := gw.conn.WriteMessage(websocket.TextMessage, bb); err != nil {
+		return errors.Wrap(err, "send message to gateway error")
+	}
+
+	return nil
+}
+
+func (b *Backend) sendRawToGateway(gatewayID lorawan.EUI64, messageType int, data []byte) error {
+	gw, err := b.gateways.get(gatewayID)
+	if err != nil {
+		return errors.Wrap(err, "get gateway error")
+	}
+
 	gw.conn.SetWriteDeadline(time.Now().Add(b.writeTimeout))
-	if err := gw.conn.WriteJSON(v); err != nil {
+	if err := gw.conn.WriteMessage(messageType, data); err != nil {
 		return errors.Wrap(err, "send message to gateway error")
 	}
 

internal/backend/basicstation/backend_test.go

@@ -12,10 +12,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
-	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/basicstation/structs"
-	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
 	"github.com/brocaar/chirpstack-api/go/v3/common"
 	"github.com/brocaar/chirpstack-api/go/v3/gw"
+	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/basicstation/structs"
+	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
 	"github.com/brocaar/lorawan"
 )
 
@@ -455,6 +455,78 @@ func (ts *BackendTestSuite) TestSendDownlinkFrame() {
 	}, df)
 }
 
+func (ts *BackendTestSuite) TestRawPacketForwarderCommand() {
+	assert := require.New(ts.T())
+	id, err := uuid.NewV4()
+	assert.NoError(err)
+
+	ts.T().Run("JSON", func(t *testing.T) {
+		assert := require.New(t)
+		pl := gw.RawPacketForwarderCommand{
+			GatewayId: []byte{1, 2, 3, 4, 5, 6, 7, 8},
+			RawId:     id[:],
+			Payload:   []byte(`{"foo": "bar"}`),
+		}
+		assert.NoError(ts.backend.RawPacketForwarderCommand(pl))
+
+		mt, msg, err := ts.wsClient.ReadMessage()
+		assert.NoError(err)
+		assert.Equal(websocket.TextMessage, mt)
+		assert.Equal(pl.Payload, msg)
+	})
+
+	ts.T().Run("Binary", func(t *testing.T) {
+		assert := require.New(t)
+		pl := gw.RawPacketForwarderCommand{
+			GatewayId: []byte{1, 2, 3, 4, 5, 6, 7, 8},
+			RawId:     id[:],
+			Payload:   []byte{0x01, 0x02, 0x03, 0x04},
+		}
+		assert.NoError(ts.backend.RawPacketForwarderCommand(pl))
+
+		mt, msg, err := ts.wsClient.ReadMessage()
+		assert.NoError(err)
+		assert.Equal(websocket.BinaryMessage, mt)
+		assert.Equal(pl.Payload, msg)
+	})
+}
+
+func (ts *BackendTestSuite) TestRawPacketForwarderEvent() {
+	ts.T().Run("Binary", func(t *testing.T) {
+		assert := require.New(t)
+
+		assert.NoError(ts.wsClient.WriteMessage(websocket.BinaryMessage, []byte{0x01, 0x02, 0x03, 0x04}))
+
+		pl := <-ts.backend.GetRawPacketForwarderEventChan()
+		assert.Equal([]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}, pl.GatewayId)
+		assert.NotNil(pl.RawId)
+		assert.Equal([]byte{0x01, 0x02, 0x03, 0x04}, pl.Payload)
+	})
+
+	ts.T().Run("JSON rmtsh", func(t *testing.T) {
+		assert := require.New(t)
+
+		jsonMsg := `{
+		  "msgtype"  : "rmtsh",
+		  "rmtsh"    : [
+			{
+			  "user"     : "foo",
+			  "started"  : true,
+			  "age"      : 1,
+			  "pid"      : 2
+			}
+		  ]
+		}`
+
+		assert.NoError(ts.wsClient.WriteMessage(websocket.TextMessage, []byte(jsonMsg)))
+
+		pl := <-ts.backend.GetRawPacketForwarderEventChan()
+		assert.Equal([]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}, pl.GatewayId)
+		assert.NotNil(pl.RawId)
+		assert.Equal([]byte(jsonMsg), pl.Payload)
+	})
+}
+
 func TestBackend(t *testing.T) {
 	suite.Run(t, new(BackendTestSuite))
 }

internal/backend/semtechudp/backend.go

@@ -14,10 +14,10 @@ import (
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/semtechudp/packets"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
 	"github.com/brocaar/chirpstack-gateway-bridge/internal/filters"
-	"github.com/brocaar/chirpstack-api/go/v3/gw"
 	"github.com/brocaar/lorawan"
 )
 
@@ -173,6 +173,12 @@ func (b *Backend) GetDisconnectChan() chan lorawan.EUI64 {
 	return b.gateways.disconnectChan
 }
 
+// GetRawPacketForwarderEventChan returns the raw packet-forwarder command channel.
+func (b *Backend) GetRawPacketForwarderEventChan() chan gw.RawPacketForwarderEvent {
+	// not provided by the Semtech packet-forwarder.
+	return nil
+}
+
 // SendDownlinkFrame sends the given downlink frame to the gateway.
 func (b *Backend) SendDownlinkFrame(frame gw.DownlinkFrame) error {
 	// mutex is needed in order to write to tokenMap
@@ -238,6 +244,11 @@ func (b *Backend) ApplyConfiguration(config gw.GatewayConfiguration) error {
 	return b.applyConfiguration(*pfConfig, config)
 }
 
+// RawPacketForwarderCommand sends the given raw command to the packet-forwarder.
+func (b *Backend) RawPacketForwarderCommand(gw.RawPacketForwarderCommand) error {
+	return errors.New("raw packet-forwarder command not implemented by Semtech packet-forwarder")
+}
+
 func (b *Backend) applyConfiguration(pfConfig pfConfiguration, config gw.GatewayConfiguration) error {
 	gwConfig, err := getGatewayConfig(config)
 	if err != nil {