You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mynewt.apache.org by GitBox <gi...@apache.org> on 2017/10/27 20:18:55 UTC
[GitHub] mkiiskila closed pull request #46: OIC over Lora
mkiiskila closed pull request #46: OIC over Lora
URL: https://github.com/apache/mynewt-newtmgr/pull/46
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/newtmgr/cli/common.go b/newtmgr/cli/common.go
index 07f5700..b45c214 100644
--- a/newtmgr/cli/common.go
+++ b/newtmgr/cli/common.go
@@ -28,6 +28,7 @@ import (
"mynewt.apache.org/newtmgr/newtmgr/bll"
"mynewt.apache.org/newtmgr/newtmgr/config"
"mynewt.apache.org/newtmgr/newtmgr/nmutil"
+ "mynewt.apache.org/newtmgr/nmxact/mtech_lora"
"mynewt.apache.org/newtmgr/nmxact/nmble"
"mynewt.apache.org/newtmgr/nmxact/nmserial"
"mynewt.apache.org/newtmgr/nmxact/sesn"
@@ -144,6 +145,10 @@ func GetXport() (xport.Xport, error) {
case config.CONN_TYPE_UDP_PLAIN, config.CONN_TYPE_UDP_OIC:
globalXport = udp.NewUdpXport()
+ case config.CONN_TYPE_MTECH_LORA_OIC:
+ cfg := mtech_lora.NewXportCfg()
+ globalXport = mtech_lora.NewLoraXport(cfg)
+
default:
return nil, util.FmtNewtError("Unknown connection type: %s (%d)",
config.ConnTypeToString(cp.Type), int(cp.Type))
@@ -233,6 +238,15 @@ func buildSesnCfg() (sesn.SesnCfg, error) {
return sc, nil
+ case config.CONN_TYPE_MTECH_LORA_OIC:
+ mc, err := config.ParseMtechLoraConnString(cp.ConnString)
+ if err != nil {
+ return sc, err
+ }
+ sc.MgmtProto = sesn.MGMT_PROTO_OMP
+ err = config.FillMtechLoraSesnCfg(mc, &sc)
+ return sc, err
+
default:
return sc, util.FmtNewtError("Unknown connection type: %s (%d)",
config.ConnTypeToString(cp.Type), int(cp.Type))
diff --git a/newtmgr/config/connprofile.go b/newtmgr/config/connprofile.go
index 0823d99..ec91353 100644
--- a/newtmgr/config/connprofile.go
+++ b/newtmgr/config/connprofile.go
@@ -59,18 +59,20 @@ const (
CONN_TYPE_BLE_OIC
CONN_TYPE_UDP_PLAIN
CONN_TYPE_UDP_OIC
+ CONN_TYPE_MTECH_LORA_OIC
)
var connTypeNameMap = map[ConnType]string{
- CONN_TYPE_SERIAL_PLAIN: "serial",
- CONN_TYPE_SERIAL_OIC: "oic_serial",
- CONN_TYPE_BLL_PLAIN: "ble",
- CONN_TYPE_BLL_OIC: "oic_ble",
- CONN_TYPE_BLE_PLAIN: "bhd",
- CONN_TYPE_BLE_OIC: "oic_bhd",
- CONN_TYPE_UDP_PLAIN: "udp",
- CONN_TYPE_UDP_OIC: "oic_udp",
- CONN_TYPE_NONE: "???",
+ CONN_TYPE_SERIAL_PLAIN: "serial",
+ CONN_TYPE_SERIAL_OIC: "oic_serial",
+ CONN_TYPE_BLL_PLAIN: "ble",
+ CONN_TYPE_BLL_OIC: "oic_ble",
+ CONN_TYPE_BLE_PLAIN: "bhd",
+ CONN_TYPE_BLE_OIC: "oic_bhd",
+ CONN_TYPE_UDP_PLAIN: "udp",
+ CONN_TYPE_UDP_OIC: "oic_udp",
+ CONN_TYPE_MTECH_LORA_OIC: "oic_mtech",
+ CONN_TYPE_NONE: "???",
}
func ConnTypeToString(ct ConnType) string {
diff --git a/newtmgr/config/mtech_lora_config.go b/newtmgr/config/mtech_lora_config.go
new file mode 100644
index 0000000..4194b04
--- /dev/null
+++ b/newtmgr/config/mtech_lora_config.go
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package config
+
+import (
+ "strconv"
+ "strings"
+
+ "mynewt.apache.org/newt/util"
+ "mynewt.apache.org/newtmgr/newtmgr/nmutil"
+ "mynewt.apache.org/newtmgr/nmxact/mtech_lora"
+ "mynewt.apache.org/newtmgr/nmxact/sesn"
+)
+
+func NewMtechLoraConfig() *mtech_lora.LoraConfig {
+ return &mtech_lora.LoraConfig{
+ Addr: "",
+ SegSz: 0,
+ }
+}
+
+func ParseMtechLoraConnString(cs string) (*mtech_lora.LoraConfig, error) {
+ mc := NewMtechLoraConfig()
+
+ if len(cs) == 0 {
+ return mc, nil
+ }
+ parts := strings.Split(cs, ",")
+ for _, p := range parts {
+ kv := strings.SplitN(p, "=", 2)
+ if len(kv) != 2 {
+ return nil, util.FmtNewtError("expected comma-separated "+
+ "key=value pairs; no '=' in: %s", p)
+ }
+
+ k := kv[0]
+ v := kv[1]
+
+ switch k {
+ case "addr":
+ mc.Addr = v
+ case "segsz":
+ var err error
+ mc.SegSz, err = strconv.Atoi(v)
+ if err != nil {
+ return mc, util.FmtNewtError("Invalid SegSz: %s", v)
+ }
+ case "confirmedtx":
+ var err error
+ mc.ConfirmedTx, err = strconv.ParseBool(v)
+ if err != nil {
+ return mc, util.FmtNewtError("Invalid confirmedtx: %s", v)
+ }
+ default:
+ return nil, util.FmtNewtError("Unrecognized key: %s", k)
+ }
+ }
+
+ return mc, nil
+}
+
+func FillMtechLoraSesnCfg(mc *mtech_lora.LoraConfig, sc *sesn.SesnCfg) error {
+ sc.Lora.Addr = mc.Addr
+ sc.Lora.SegSz = mc.SegSz
+ sc.Lora.ConfirmedTx = mc.ConfirmedTx
+ if nmutil.DeviceName != "" {
+ sc.Lora.Addr = nmutil.DeviceName
+ }
+ return nil
+}
diff --git a/nmxact/lora/lora_coap.go b/nmxact/lora/lora_coap.go
new file mode 100644
index 0000000..fbd6206
--- /dev/null
+++ b/nmxact/lora/lora_coap.go
@@ -0,0 +1,31 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package lora
+
+type CoapLoraFragStart struct {
+ FragNum uint8
+ Crc uint16
+}
+
+type CoapLoraFrag struct {
+ FragNum uint8
+}
+
+const COAP_LORA_LAST_FRAG uint8 = 0x80
diff --git a/nmxact/mtech_lora/listen.go b/nmxact/mtech_lora/listen.go
new file mode 100644
index 0000000..0bc6976
--- /dev/null
+++ b/nmxact/mtech_lora/listen.go
@@ -0,0 +1,201 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package mtech_lora
+
+import (
+ "bytes"
+ "fmt"
+ "time"
+
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
+)
+
+type ListenerKey struct {
+ Target string
+ Type string
+}
+
+func TgtKey(tgt string, msgType string) ListenerKey {
+ return ListenerKey{
+ Target: tgt,
+ Type: msgType,
+ }
+}
+
+func TypeKey(msgType string) ListenerKey {
+ return ListenerKey{
+ Target: "",
+ Type: msgType,
+ }
+}
+
+type Listener struct {
+ MsgChan chan []byte
+ ErrChan chan error
+ TmoChan chan time.Time
+ Acked bool
+
+ Data *bytes.Buffer
+ NextFrag uint8
+ Crc uint16
+
+ timer *time.Timer
+}
+
+func NewListener() *Listener {
+ return &Listener{
+ MsgChan: make(chan []byte, 16),
+ ErrChan: make(chan error, 1),
+ TmoChan: make(chan time.Time, 1),
+
+ Data: bytes.NewBuffer([]byte{}),
+ }
+}
+
+func (ll *Listener) AfterTimeout(tmo time.Duration) <-chan time.Time {
+ fn := func() {
+ if !ll.Acked {
+ ll.TmoChan <- time.Now()
+ }
+ }
+ ll.timer = time.AfterFunc(tmo, fn)
+ return ll.TmoChan
+}
+
+func (ll *Listener) Close() {
+ // This provokes a race condition. The timer may get initialized at any
+ // time.
+ if ll.timer != nil {
+ ll.timer.Stop()
+ }
+
+ // Mark the command as acked in case the race condition mentioned above
+ // occurred. If the timer goes off, nothing will happen.
+ ll.Acked = true
+
+ close(ll.MsgChan)
+ for {
+ if _, ok := <-ll.MsgChan; !ok {
+ break
+ }
+ }
+
+ close(ll.ErrChan)
+ for {
+ if _, ok := <-ll.ErrChan; !ok {
+ break
+ }
+ }
+
+ close(ll.TmoChan)
+ for {
+ if _, ok := <-ll.TmoChan; !ok {
+ break
+ }
+ }
+}
+
+// Not thread safe.
+type ListenerMap struct {
+ k2l map[ListenerKey]*Listener
+ l2k map[*Listener]ListenerKey
+}
+
+func NewListenerMap() *ListenerMap {
+ return &ListenerMap{
+ k2l: map[ListenerKey]*Listener{},
+ l2k: map[*Listener]ListenerKey{},
+ }
+}
+
+func (lm *ListenerMap) FindListener(tgt string, msgType string) (
+ ListenerKey, *Listener) {
+
+ var key ListenerKey
+
+ key = TgtKey(tgt, msgType)
+ if listener := lm.k2l[key]; listener != nil {
+ return key, listener
+ }
+
+ key = TypeKey(msgType)
+ if listener := lm.k2l[key]; listener != nil {
+ return key, listener
+ }
+
+ return key, nil
+}
+
+func (lm *ListenerMap) AddListener(key ListenerKey, listener *Listener) error {
+ if _, ok := lm.k2l[key]; ok {
+ nmxutil.Assert(false)
+ return fmt.Errorf("Duplicate Lora listener: %#v", key)
+ }
+
+ if _, ok := lm.l2k[listener]; ok {
+ nmxutil.Assert(false)
+ return fmt.Errorf("Duplicate Lora listener: %#v", key)
+ }
+
+ lm.k2l[key] = listener
+ lm.l2k[listener] = key
+
+ return nil
+}
+
+func (lm *ListenerMap) deleteListener(key ListenerKey, listener *Listener) {
+ nmxutil.Assert(lm.k2l[key] == listener)
+ nmxutil.Assert(lm.l2k[listener] == key)
+ delete(lm.k2l, key)
+ delete(lm.l2k, listener)
+}
+
+func (lm *ListenerMap) RemoveListener(listener *Listener) *ListenerKey {
+ key, ok := lm.l2k[listener]
+ if !ok {
+ return nil
+ }
+
+ lm.deleteListener(key, listener)
+ return &key
+}
+
+func (lm *ListenerMap) RemoveKey(key ListenerKey) *Listener {
+ listener := lm.k2l[key]
+ if listener == nil {
+ return nil
+ }
+
+ lm.deleteListener(key, listener)
+ return listener
+}
+
+func (lm *ListenerMap) ExtractAll() []*Listener {
+ listeners := make([]*Listener, 0, len(lm.l2k))
+
+ for listener, _ := range lm.l2k {
+ listeners = append(listeners, listener)
+ }
+
+ lm.k2l = map[ListenerKey]*Listener{}
+ lm.l2k = map[*Listener]ListenerKey{}
+
+ return listeners
+}
diff --git a/nmxact/mtech_lora/mtech_lora_sesn.go b/nmxact/mtech_lora/mtech_lora_sesn.go
new file mode 100644
index 0000000..a4f94b5
--- /dev/null
+++ b/nmxact/mtech_lora/mtech_lora_sesn.go
@@ -0,0 +1,257 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package mtech_lora
+
+import (
+ "bytes"
+ "encoding/base64"
+ "encoding/binary"
+ "fmt"
+ "strings"
+ "sync"
+
+ "github.com/joaojeronimo/go-crc16"
+ "github.com/runtimeco/go-coap"
+ "github.com/ugorji/go/codec"
+
+ "mynewt.apache.org/newtmgr/nmxact/lora"
+ "mynewt.apache.org/newtmgr/nmxact/mgmt"
+ "mynewt.apache.org/newtmgr/nmxact/nmp"
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
+ "mynewt.apache.org/newtmgr/nmxact/omp"
+ "mynewt.apache.org/newtmgr/nmxact/sesn"
+)
+
+type LoraSesn struct {
+ cfg sesn.SesnCfg
+ txvr *mgmt.Transceiver
+ isOpen bool
+ xport *LoraXport
+ listener *Listener
+ wg sync.WaitGroup
+ stopChan chan struct{}
+}
+
+type mtechLoraTx struct {
+ Port uint16 `json:"port"`
+ Data string `json:"data"`
+ Ack bool `json:"ack"`
+}
+
+func normalizeAddr(addr string) (string, error) {
+ a := strings.Replace(addr, ":", "-", -1)
+ // XXX check that there's 8 components, 2 chars each, which are in [0-9,a-f]
+ return a, nil
+}
+
+func NewLoraSesn(cfg sesn.SesnCfg, lx *LoraXport) (*LoraSesn, error) {
+ addr, err := normalizeAddr(cfg.Lora.Addr)
+ if err != nil {
+ return nil, fmt.Errorf("Invalid Lora address %s\n", cfg.Lora.Addr)
+ }
+ cfg.Lora.Addr = addr
+ s := &LoraSesn{
+ cfg: cfg,
+ xport: lx,
+ }
+
+ return s, nil
+}
+
+func (s *LoraSesn) Open() error {
+ if s.isOpen == true {
+ return nmxutil.NewSesnAlreadyOpenError(
+ "Attempt to open an already-open Lora session")
+ }
+
+ key := TgtKey(s.cfg.Lora.Addr, "rx")
+ s.xport.Lock()
+
+ txvr, err := mgmt.NewTransceiver(false, s.cfg.MgmtProto, 3)
+ if err != nil {
+ return err
+ }
+ s.txvr = txvr
+ s.stopChan = make(chan struct{})
+ s.listener = NewListener()
+
+ err = s.xport.listenMap.AddListener(key, s.listener)
+ if err != nil {
+ s.txvr.Stop()
+ return err
+ }
+ s.xport.Unlock()
+
+ s.wg.Add(1)
+ go func() {
+ defer s.wg.Done()
+ defer s.xport.listenMap.RemoveListener(s.listener)
+
+ for {
+ select {
+ case msg, ok := <-s.listener.MsgChan:
+ if ok {
+ s.txvr.DispatchCoap(msg)
+ }
+ case <-s.stopChan:
+ return
+ }
+ }
+ }()
+ s.isOpen = true
+ return nil
+}
+
+func (s *LoraSesn) Close() error {
+ if s.isOpen == false {
+ return nmxutil.NewSesnClosedError(
+ "Attempt to close an unopened Lora session")
+ }
+
+ s.isOpen = false
+ s.txvr.ErrorAll(fmt.Errorf("manual close"))
+ s.txvr.Stop()
+ close(s.stopChan)
+ s.listener.Close()
+ s.wg.Wait()
+ s.stopChan = nil
+ s.txvr = nil
+
+ return nil
+}
+
+func (s *LoraSesn) IsOpen() bool {
+ return s.isOpen
+}
+
+func (s *LoraSesn) MtuIn() int {
+ return MAX_PACKET_SIZE - omp.OMP_MSG_OVERHEAD - nmp.NMP_HDR_SIZE
+}
+
+func (s *LoraSesn) MtuOut() int {
+ return MAX_PACKET_SIZE - omp.OMP_MSG_OVERHEAD - nmp.NMP_HDR_SIZE
+}
+
+func (s *LoraSesn) sendFragments(b []byte) error {
+ segSz := s.xport.minMtu()
+ if segSz < s.cfg.Lora.SegSz {
+ segSz = s.cfg.Lora.SegSz
+ }
+ crc := crc16.Crc16(b)
+ idx := 0
+ for off := 0; off < len(b); {
+ var seg bytes.Buffer
+ var blkLen int
+ if off == 0 {
+ hdr := lora.CoapLoraFragStart{
+ FragNum: 0,
+ Crc: crc,
+ }
+ blkLen = segSz - 4
+ if blkLen >= len(b) {
+ blkLen = len(b)
+ hdr.FragNum |= lora.COAP_LORA_LAST_FRAG
+ }
+ binary.Write(&seg, binary.LittleEndian, hdr)
+ seg.Write(b[0:blkLen])
+ } else {
+ hdr := lora.CoapLoraFrag{
+ FragNum: uint8(idx),
+ }
+ blkLen = segSz - 1
+ if blkLen >= len(b)-off {
+ blkLen = len(b) - off
+ hdr.FragNum |= lora.COAP_LORA_LAST_FRAG
+ }
+ binary.Write(&seg, binary.LittleEndian, hdr)
+ seg.Write(b[off : off+blkLen])
+ }
+ off += blkLen
+ idx++
+
+ seg64 := make([]byte, base64.StdEncoding.EncodedLen(len(seg.Bytes())))
+ base64.StdEncoding.Encode(seg64, seg.Bytes())
+
+ msg := mtechLoraTx{
+ Port: OIC_LORA_PORT,
+ Ack: s.cfg.Lora.ConfirmedTx,
+ Data: string(seg64),
+ }
+
+ payload := []byte{}
+ enc := codec.NewEncoderBytes(&payload, new(codec.JsonHandle))
+ enc.Encode(msg)
+
+ var outData bytes.Buffer
+
+ outData.Write([]byte(fmt.Sprintf("lora/%s/down %s\n",
+ s.cfg.Lora.Addr, payload)))
+ err := s.xport.Tx(outData.Bytes())
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (s *LoraSesn) TxNmpOnce(m *nmp.NmpMsg, opt sesn.TxOptions) (
+ nmp.NmpRsp, error) {
+
+ if !s.IsOpen() {
+ return nil, fmt.Errorf("Attempt to transmit over closed Lora session")
+ }
+
+ txFunc := func(b []byte) error {
+ return s.sendFragments(b)
+ }
+ return s.txvr.TxNmp(txFunc, m, s.MtuOut(), opt.Timeout)
+}
+
+func (s *LoraSesn) AbortRx(seq uint8) error {
+ s.txvr.ErrorAll(fmt.Errorf("Rx aborted"))
+ return nil
+}
+
+func (s *LoraSesn) TxCoapOnce(m coap.Message, resType sesn.ResourceType,
+ opt sesn.TxOptions) (coap.COAPCode, []byte, error) {
+
+ if !s.IsOpen() {
+ return 0, nil, fmt.Errorf("Attempt to transmit over closed Lora session")
+ }
+ txFunc := func(b []byte) error {
+ return s.sendFragments(b)
+ }
+ rsp, err := s.txvr.TxOic(txFunc, m, s.MtuOut(), opt.Timeout)
+ if err != nil {
+ return 0, nil, err
+ } else if rsp == nil {
+ return 0, nil, nil
+ } else {
+ return rsp.Code(), rsp.Payload(), nil
+ }
+}
+
+func (s *LoraSesn) MgmtProto() sesn.MgmtProto {
+ return s.cfg.MgmtProto
+}
+
+func (s *LoraSesn) CoapIsTcp() bool {
+ return false
+}
diff --git a/nmxact/mtech_lora/mtech_lora_xport.go b/nmxact/mtech_lora/mtech_lora_xport.go
new file mode 100644
index 0000000..082faf8
--- /dev/null
+++ b/nmxact/mtech_lora/mtech_lora_xport.go
@@ -0,0 +1,277 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package mtech_lora
+
+import (
+ "bytes"
+ "encoding/base64"
+ "encoding/binary"
+ "encoding/hex"
+ "fmt"
+ "net"
+ "strings"
+ "sync"
+
+ log "github.com/Sirupsen/logrus"
+ "github.com/ugorji/go/codec"
+
+ "mynewt.apache.org/newtmgr/nmxact/lora"
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
+ "mynewt.apache.org/newtmgr/nmxact/sesn"
+)
+
+type LoraConfig struct {
+ Addr string
+ SegSz int
+ ConfirmedTx bool
+}
+
+type LoraJoinedCb func(dev LoraConfig)
+
+type LoraXport struct {
+ sync.Mutex
+ cfg *LoraXportCfg
+ started bool
+ rxConn *net.UDPConn
+ txConn *net.UDPConn
+ listenMap *ListenerMap
+ exitChan chan int
+ joinCb LoraJoinedCb
+}
+
+type LoraXportCfg struct {
+}
+
+type LoraData struct {
+ Data string `codec:"data"`
+ Port int `codec:"port"`
+}
+
+const MAX_PACKET_SIZE = 2048
+const UDP_RX_PORT = 1784
+const UDP_TX_PORT = 1786
+const OIC_LORA_PORT = 0xbb
+
+func NewXportCfg() *LoraXportCfg {
+ return &LoraXportCfg{}
+}
+
+func NewLoraXport(cfg *LoraXportCfg) *LoraXport {
+ return &LoraXport{
+ cfg: cfg,
+ listenMap: NewListenerMap(),
+ }
+}
+
+func (lx *LoraXport) minMtu() int {
+ return 33
+}
+
+func (lx *LoraXport) BuildSesn(cfg sesn.SesnCfg) (sesn.Sesn, error) {
+ if cfg.Lora.Addr == "" {
+ return nil, fmt.Errorf("Need an address of endpoint")
+ }
+ return NewLoraSesn(cfg, lx)
+}
+
+func (lx *LoraXport) reass(dev string, data []byte) {
+ lx.Lock()
+ defer lx.Unlock()
+
+ _, l := lx.listenMap.FindListener(dev, "rx")
+ if l == nil {
+ log.Debugf("No LoRa listener for %s", dev)
+ return
+ }
+
+ str := hex.Dump(data)
+
+ log.Debugf("Raw data: %s", str)
+ bufR := bytes.NewReader(data)
+
+ var frag lora.CoapLoraFrag
+ var sFrag lora.CoapLoraFragStart
+
+ err := binary.Read(bufR, binary.LittleEndian, &frag)
+ if err != nil {
+ log.Debugf("Can't read header")
+ return
+ }
+
+ fragNum := frag.FragNum &^ lora.COAP_LORA_LAST_FRAG
+ if l.Data.Len() == 0 {
+ if fragNum != 0 {
+ log.Debugf("frag num != 0 with empty queue")
+ return
+ }
+ bufR.Seek(0, 0)
+ err = binary.Read(bufR, binary.LittleEndian, &sFrag)
+ if err != nil {
+ log.Debugf("Can't read in start header")
+ return
+ }
+ l.Crc = sFrag.Crc
+ l.Data.Write(data[3:])
+ l.NextFrag = 1
+ } else {
+ if fragNum != l.NextFrag {
+ log.Debugf("Frag out of sequence")
+ l.Data.Reset()
+ return
+ }
+ l.Data.Write(data[1:])
+ l.NextFrag++
+ }
+
+ if (frag.FragNum & lora.COAP_LORA_LAST_FRAG) != 0 {
+ l.MsgChan <- l.Data.Bytes()
+ l.Data.Reset()
+ }
+}
+
+/*
+ * lora/00-13-50-04-04-50-13-00/up
+ */
+func (lx *LoraXport) processData(data string) {
+ if strings.HasPrefix(data, "lora/") == false {
+ return
+ }
+ splitMsg := strings.Fields(data)
+ if len(splitMsg) == 0 {
+ return
+ }
+ splitHdr := strings.Split(splitMsg[0], "/")
+ if len(splitHdr) != 3 {
+ return
+ }
+ switch splitHdr[2] {
+ case "joined":
+ log.Debugf("loraxport rx: %s", data)
+ log.Debugf("%s joined", splitHdr[1])
+ lx.reportJoin(splitHdr[1])
+ case "up":
+ var msg LoraData
+
+ log.Debugf("loraxport rx: %s", data)
+ pload := []byte(splitMsg[1])
+ err := codec.NewDecoderBytes(pload, new(codec.JsonHandle)).Decode(&msg)
+ if err != nil {
+ log.Debugf("loraxport rx: error decoding json: %v", err)
+ return
+ }
+
+ if msg.Port != OIC_LORA_PORT {
+ log.Debugf("loraxport rx: ignoring data to port %d", msg.Port)
+ return
+ }
+ dec, err := base64.StdEncoding.DecodeString(msg.Data)
+ if err != nil {
+ log.Debugf("loraxport rx: error decoding base64: %v", err)
+ return
+ }
+ lx.reass(splitHdr[1], dec)
+ }
+}
+
+func (lx *LoraXport) Start() error {
+ if lx.started {
+ return nmxutil.NewXportError("Lora xport started twice")
+ }
+
+ addr, err := net.ResolveUDPAddr("udp", "127.0.0.1:1784")
+ if err != nil {
+ return fmt.Errorf("Failure resolving name for UDP session: %s",
+ err.Error())
+ }
+
+ // XXX need so_reuseport
+ conn, err := net.ListenUDP("udp", addr)
+ if err != nil {
+ return fmt.Errorf("Failed to open RX to lora-network-server %s", addr)
+ }
+ lx.rxConn = conn
+
+ addr, err = net.ResolveUDPAddr("udp", "127.0.0.1:1786")
+ if err != nil {
+ return fmt.Errorf("Failure resolving name for UDP session: %s",
+ err.Error())
+ }
+ conn, err = net.DialUDP("udp", nil, addr)
+ if err != nil {
+ lx.rxConn.Close()
+ lx.rxConn = nil
+ return fmt.Errorf("Failed to open TX to lora-network-server")
+ }
+ lx.txConn = conn
+
+ lx.started = true
+
+ go func() {
+ data := make([]byte, MAX_PACKET_SIZE*4/3+512)
+ for {
+ nr, _, err := lx.rxConn.ReadFromUDP(data)
+ if err != nil {
+ return
+ }
+ lx.processData(string(data[0:nr]))
+ }
+ }()
+
+ return nil
+}
+
+func (lx *LoraXport) reportJoin(dev string) {
+ lx.Lock()
+ if lx.joinCb != nil {
+ dev := LoraConfig{
+ Addr: dev,
+ }
+ lx.Unlock()
+ lx.joinCb(dev)
+ } else {
+ lx.Unlock()
+ }
+}
+
+func (lx *LoraXport) SetJoinCb(joinCb LoraJoinedCb) {
+ lx.Lock()
+ defer lx.Unlock()
+
+ lx.joinCb = joinCb
+}
+
+func (lx *LoraXport) Stop() error {
+ if !lx.started {
+ return nmxutil.NewXportError("Lora xport stopped twice")
+ }
+ lx.rxConn.Close()
+ lx.rxConn = nil
+ lx.txConn.Close()
+ lx.txConn = nil
+ lx.started = false
+ lx.joinCb = nil
+ return nil
+}
+
+func (lx *LoraXport) Tx(bytes []byte) error {
+ log.Debugf("loraxport tx: %s", bytes)
+ _, err := lx.txConn.Write(bytes)
+ return err
+}
diff --git a/nmxact/sesn/sesn_cfg.go b/nmxact/sesn/sesn_cfg.go
index 9ca36f2..422f5dc 100644
--- a/nmxact/sesn/sesn_cfg.go
+++ b/nmxact/sesn/sesn_cfg.go
@@ -85,6 +85,12 @@ type SesnCfgBle struct {
Central SesnCfgBleCentral
}
+type SesnCfgLora struct {
+ Addr string
+ SegSz int
+ ConfirmedTx bool
+}
+
type SesnCfg struct {
// General configuration.
MgmtProto MgmtProto
@@ -92,7 +98,8 @@ type SesnCfg struct {
OnCloseCb OnCloseFn
// Transport-specific configuration.
- Ble SesnCfgBle
+ Ble SesnCfgBle
+ Lora SesnCfgLora
}
func NewSesnCfg() SesnCfg {
@@ -110,5 +117,8 @@ func NewSesnCfg() SesnCfg {
ConnTimeout: 10 * time.Second,
},
},
+ Lora: SesnCfgLora{
+ ConfirmedTx: false,
+ },
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services