Implement TX ACK and RX time / tmms field.

Closes #19.

Author: brocaar

backend/mqttpubsub/backend.go

@@ -3,16 +3,16 @@ package mqttpubsub
 import (
 	"crypto/tls"
 	"crypto/x509"
-	"io/ioutil"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"sync"
 	"time"
 
-	log "github.com/sirupsen/logrus"
 	"github.com/brocaar/loraserver/api/gw"
 	"github.com/brocaar/lorawan"
 	"github.com/eclipse/paho.mqtt.golang"
+	log "github.com/sirupsen/logrus"
 )
 
 // Backend implements a MQTT pub-sub backend.
@@ -125,6 +125,12 @@ func (b *Backend) PublishGatewayStats(mac lorawan.EUI64, stats gw.GatewayStatsPa
 	return b.publish(topic, stats)
 }
 
+// PublishGatewayTXAck publishes a TX ack to the MQTT broker.
+func (b *Backend) PublishGatewayTXAck(mac lorawan.EUI64, ack gw.TXAck) error {
+	topic := fmt.Sprintf("gateway/%s/ack", mac.String())
+	return b.publish(topic, ack)
+}
+
 func (b *Backend) publish(topic string, v interface{}) error {
 	bytes, err := json.Marshal(v)
 	if err != nil {

backend/mqttpubsub/backend_test.go

@@ -39,10 +39,11 @@ func TestBackend(t *testing.T) {
 				So(token.Error(), ShouldBeNil)
 
 				Convey("When publishing a RXPacket", func() {
+					now := time.Now().UTC()
 					rxPacket := gw.RXPacketBytes{
 						RXInfo: gw.RXInfo{
 							MAC:  [8]byte{1, 2, 3, 4, 5, 6, 7, 8},
-							Time: time.Now().UTC(),
+							Time: &now,
 						},
 						PHYPayload: []byte{1, 2, 3, 4},
 					}

cmd/lora-gateway-bridge/main.go

@@ -8,11 +8,11 @@ import (
 	"syscall"
 	"time"
 
-	log "github.com/sirupsen/logrus"
 	"github.com/brocaar/lora-gateway-bridge/backend/mqttpubsub"
 	"github.com/brocaar/lora-gateway-bridge/gateway"
 	"github.com/brocaar/lorawan"
 	"github.com/codegangsta/cli"
+	log "github.com/sirupsen/logrus"
 )
 
 var version string // set by the compiler
@@ -76,6 +76,14 @@ func run(c *cli.Context) error {
 		}
 	}()
 
+	go func() {
+		for txAck := range gw.TXAckChan() {
+			if err := pubsub.PublishGatewayTXAck(txAck.MAC, txAck); err != nil {
+				log.Errorf("could not publish TXAck: %s", err)
+			}
+		}
+	}()
+
 	sigChan := make(chan os.Signal)
 	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
 	log.WithField("signal", <-sigChan).Info("signal received")

docs/config.toml

@@ -32,4 +32,4 @@ googleAnalytics = "UA-3512995-9"
     weight = 4
 
 [params]
-    version = "2.1.6"
+    version = "2.2.0"

docs/content/overview/changelog.md

@@ -8,6 +8,22 @@ menu:
 
 ## Changelog
 
+### 2.2.0
+
+**Features:**
+
+* LoRa Gateway Bridge now publishes TX acknowledgement messages over MQTT.
+  See [MQTT topics](https://docs.loraserver.io/lora-gateway-bridge/use/data/).
+
+* TX (GPS) time field is now implemented to transmit at given timestamp
+  (only possible when the gateway has a GPS time-source).
+
+**Bugfixes:**
+
+* Without GPS time-source, the gateway would use `0001-01-01T00:00:00Z`
+  as RX `time`. The `time` field is now omitted when unavailable.
+
+
 ### 2.1.6
 
 **Features:**
@@ -28,7 +44,6 @@ menu:
   re-connect the protocol version would be downgraded
   ([paho.mqtt.golang#116](https://github.com/eclipse/paho.mqtt.golang/issues/116)).
 
-
 ### 2.1.4
 
 * Retry connecting to MQTT broker on startup (thanks @jcampanell-cablelabs).

docs/content/use/data.md

@@ -64,7 +64,7 @@ Topic for received packets (from nodes). Example payload:
         "rfChain": 1,
         "rssi": -57,
         "size": 23,
-        "time": "0001-01-01T00:00:00Z",
+        "time": "2017-12-11T10:14:54.619571Z",         // timestamp (only set when the gateway has a GPS time-source)
         "timestamp": 2074240683                        // gateway internal timestamp (32 bit) with microsecond precision
     }
 }
@@ -77,6 +77,7 @@ Example payload:
 
 ```json
 {
+    "token": 65535,  // random token (uint16), used for acknowledgements
     "phyPayload": "IKu70cumKom7BREUFrxlHtM=",
     "txInfo": {
         "board": 0,
@@ -91,11 +92,37 @@ Example payload:
         "immediately": false,
         "mac": "1dee08d0b691d149",
         "power": 14,
-        "timestamp": 2079240683
+        "timestamp": 2079240683,               // gateway internal timestamp for transmission -OR-
+        "time": "2017-12-11T10:14:54.619571Z"  // timestamp for transmission (only when the gateway has a GPS time-source)
     }
 }
 ```
 
+When `immediately` is set to `false`, either the `timestamp` **or** the
+`time` field must be present to tell the gateway at what (internal) time
+the frame must be transmitted.
+
 Optionally, the field `iPol` (type `bool`) can be used to control the
 LoRa modulation polarization inversion. When left blank (`null`), the default
-will be used (which is `true` for downlink LoRa modulation.
+will be used (which is `true` for downlink LoRa modulation.
+
+### gateway/[mac]/ack
+
+Topic for received TX acknowledgements (or TX errors). Example payload:
+
+```json
+{
+    "token": 65535,              // same token used during transmission
+    "error": "COLLISION_PACKET"  // not set in case of acknowledgement
+}
+```
+
+Possible error values are:
+
+* `TOO_LATE`: Rejected because it was already too late to program this packet for downlink
+* `TOO_EARLY`: Rejected because downlink packet timestamp is too much in advance
+* `COLLISION_PACKET`: Rejected because there was already a packet programmed in requested timeframe
+* `COLLISION_BEACON`: Rejected because there was already a beacon planned in requested timeframe
+* `TX_FREQ`: Rejected because requested frequency is not supported by TX RF chain
+* `TX_POWER`: Rejected because requested power is not supported by gateway
+* `GPS_UNLOCKED`: Rejected because GPS is unlocked, so GPS timestamp cannot be used

gateway/backend.go

@@ -81,6 +81,7 @@ func (c *gateways) cleanup() error {
 // Backend implements a Semtech gateway backend.
 type Backend struct {
 	conn         *net.UDPConn
+	txAckChan    chan gw.TXAck
 	rxChan       chan gw.RXPacketBytes
 	statsChan    chan gw.GatewayStatsPacket
 	udpSendChan  chan udpPacket
@@ -105,6 +106,7 @@ func NewBackend(bind string, onNew func(lorawan.EUI64) error, onDelete func(lora
 	b := &Backend{
 		skipCRCCheck: skipCRCCheck,
 		conn:         conn,
+		txAckChan:    make(chan gw.TXAck),
 		rxChan:       make(chan gw.RXPacketBytes),
 		statsChan:    make(chan gw.GatewayStatsPacket),
 		udpSendChan:  make(chan udpPacket),
@@ -163,11 +165,17 @@ func (b *Backend) RXPacketChan() chan gw.RXPacketBytes {
 	return b.rxChan
 }
 
-// StatsChan returns the channel containg the received gateway stats.
+// StatsChan returns the channel containing the received gateway stats.
 func (b *Backend) StatsChan() chan gw.GatewayStatsPacket {
 	return b.statsChan
 }
 
+// TXAckChan returns the channel containing the TX acknowledgements
+// (or errors).
+func (b *Backend) TXAckChan() chan gw.TXAck {
+	return b.txAckChan
+}
+
 // Send sends the given packet to the gateway.
 func (b *Backend) Send(txPacket gw.TXPacketBytes) error {
 	gw, err := b.gateways.get(txPacket.TXInfo.MAC)
@@ -179,6 +187,7 @@ func (b *Backend) Send(txPacket gw.TXPacketBytes) error {
 		return err
 	}
 	pullResp := PullRespPacket{
+		RandomToken:     txPacket.Token,
 		ProtocolVersion: gw.protocolVersion,
 		Payload: PullRespPayload{
 			TXPK: txpk,
@@ -365,23 +374,10 @@ func (b *Backend) handleTXACK(addr *net.UDPAddr, data []byte) error {
 	if err := p.UnmarshalBinary(data); err != nil {
 		return err
 	}
-	var errBool bool
 
-	logFields := log.Fields{
-		"mac":          p.GatewayMAC,
-		"random_token": p.RandomToken,
-	}
 	if p.Payload != nil {
-		if p.Payload.TXPKACK.Error != "NONE" {
-			errBool = true
-		}
-		logFields["error"] = p.Payload.TXPKACK.Error
-	}
-
-	if errBool {
-		log.WithFields(logFields).Error("gateway: tx ack received")
-	} else {
-		log.WithFields(logFields).Info("gateway: tx ack received")
+		txAck := newTXAckFromTXPKACK(p.GatewayMAC, p.RandomToken, p.Payload.TXPKACK)
+		b.txAckChan <- txAck
 	}
 
 	return nil
@@ -428,7 +424,6 @@ func newRXPacketsFromRXPK(mac lorawan.EUI64, rxpk RXPK) ([]gw.RXPacketBytes, err
 		PHYPayload: b,
 		RXInfo: gw.RXInfo{
 			MAC:       mac,
-			Time:      time.Time(rxpk.Time),
 			Timestamp: rxpk.Tmst,
 			Frequency: int(rxpk.Freq * 1000000),
 			Channel:   int(rxpk.Chan),
@@ -443,6 +438,11 @@ func newRXPacketsFromRXPK(mac lorawan.EUI64, rxpk RXPK) ([]gw.RXPacketBytes, err
 		},
 	}
 
+	if rxpk.Time != nil {
+		ts := time.Time(*rxpk.Time)
+		rxPacket.RXInfo.Time = &ts
+	}
+
 	if len(rxpk.RSig) == 0 {
 		rxPackets = append(rxPackets, rxPacket)
 	}
@@ -476,6 +476,11 @@ func newTXPKFromTXPacket(txPacket gw.TXPacketBytes) (TXPK, error) {
 		Brd:  uint8(txPacket.TXInfo.Board),
 	}
 
+	if txPacket.TXInfo.Time != nil {
+		ct := CompactTime(*txPacket.TXInfo.Time)
+		txpk.Tmms = &ct
+	}
+
 	if txPacket.TXInfo.DataRate.Modulation == band.FSKModulation {
 		txpk.FDev = uint16(txPacket.TXInfo.DataRate.BitRate / 2)
 	}
@@ -491,6 +496,19 @@ func newTXPKFromTXPacket(txPacket gw.TXPacketBytes) (TXPK, error) {
 	return txpk, nil
 }
 
+func newTXAckFromTXPKACK(mac lorawan.EUI64, token uint16, ack TXPKACK) gw.TXAck {
+	var err string
+	if ack.Error != "NONE" {
+		err = ack.Error
+	}
+
+	return gw.TXAck{
+		MAC:   mac,
+		Token: token,
+		Error: err,
+	}
+}
+
 func newDataRateFromDatR(d DatR) (band.DataRate, error) {
 	var dr band.DataRate