You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mynewt.apache.org by cc...@apache.org on 2017/07/16 20:27:54 UTC
[mynewt-newtmgr] branch master updated (16457af -> 86d517d)
This is an automated email from the ASF dual-hosted git repository.
ccollins pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git.
from 16457af newtmgr - Revendor.
new 6ee575a nmxact - Simplify listener lookup.
new 86d517d newtmgr - revendor
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
newtmgr/Godeps/Godeps.json | 48 +++---
.../newtmgr/nmxact/nmble/ble_fsm.go | 56 +++----
.../newtmgr/nmxact/nmble/ble_util.go | 69 +++-----
.../newtmgr/nmxact/nmble/ble_xport.go | 54 ++----
.../newtmgr/nmxact/nmble/discover.go | 30 +---
.../newtmgr/nmxact/nmble/dispatch.go | 182 +++++----------------
.../newtmgr/nmxact/nmble/listen.go | 179 ++++++++++++++++++++
.../newtmgr/nmxact/nmble/receiver.go | 78 ++++-----
.../newtmgr/nmxact/nmp/dispatch.go | 6 +-
.../newtmgr/nmxact/nmxutil/nmxutil.go | 12 +-
.../newtmgr/nmxact/oic/dispatch.go | 39 +++--
.../newtmgr/nmxact/omp/dispatch.go | 56 +++----
nmxact/nmble/ble_fsm.go | 56 +++----
nmxact/nmble/ble_util.go | 69 +++-----
nmxact/nmble/ble_xport.go | 54 ++----
nmxact/nmble/discover.go | 30 +---
nmxact/nmble/dispatch.go | 182 +++++----------------
nmxact/nmble/listen.go | 179 ++++++++++++++++++++
nmxact/nmble/receiver.go | 78 ++++-----
nmxact/nmp/dispatch.go | 6 +-
nmxact/nmxutil/nmxutil.go | 12 +-
nmxact/oic/dispatch.go | 39 +++--
nmxact/omp/dispatch.go | 56 +++----
23 files changed, 782 insertions(+), 788 deletions(-)
create mode 100644 newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/listen.go
create mode 100644 nmxact/nmble/listen.go
--
To stop receiving notification emails like this one, please contact
['"commits@mynewt.apache.org" <co...@mynewt.apache.org>'].
[mynewt-newtmgr] 01/02: nmxact - Simplify listener lookup.
Posted by cc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ccollins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git
commit 6ee575a954bbab91f97cf48eb92d76482ae76c5a
Author: Christopher Collins <cc...@apache.org>
AuthorDate: Sun Jul 16 13:25:49 2017 -0700
nmxact - Simplify listener lookup.
---
nmxact/nmble/ble_fsm.go | 56 +++++++-------
nmxact/nmble/ble_util.go | 69 +++++-------------
nmxact/nmble/ble_xport.go | 54 +++++---------
nmxact/nmble/discover.go | 30 +++-----
nmxact/nmble/dispatch.go | 182 ++++++++++------------------------------------
nmxact/nmble/listen.go | 179 +++++++++++++++++++++++++++++++++++++++++++++
nmxact/nmble/receiver.go | 78 ++++++++------------
nmxact/nmp/dispatch.go | 6 +-
nmxact/nmxutil/nmxutil.go | 12 +--
nmxact/oic/dispatch.go | 39 ++++++----
nmxact/omp/dispatch.go | 56 +++++++-------
11 files changed, 379 insertions(+), 382 deletions(-)
diff --git a/nmxact/nmble/ble_fsm.go b/nmxact/nmble/ble_fsm.go
index 6721eb0..8d80bfd 100644
--- a/nmxact/nmble/ble_fsm.go
+++ b/nmxact/nmble/ble_fsm.go
@@ -203,7 +203,7 @@ func (bf *BleFsm) eventListen(bl *Listener, seq BleSeq) error {
go func() {
defer bf.wg.Done()
- defer bf.rxer.RemoveSeqListener("connect", seq)
+ defer bf.rxer.RemoveListener("connect", bl)
for {
select {
@@ -254,14 +254,8 @@ func (bf *BleFsm) eventListen(bl *Listener, seq BleSeq) error {
}
func (bf *BleFsm) nmpRspListen() error {
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_NOTIFY_RX_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: int(bf.connHandle),
- }
-
- bl, err := bf.rxer.AddBaseListener("nmp-rsp", base)
+ key := TchKey(MSG_TYPE_NOTIFY_RX_EVT, int(bf.connHandle))
+ bl, err := bf.rxer.AddListener("nmp-rsp", key)
if err != nil {
return err
}
@@ -270,12 +264,12 @@ func (bf *BleFsm) nmpRspListen() error {
go func() {
defer bf.wg.Done()
- defer bf.rxer.RemoveBaseListener("nmp-rsp", base)
+ defer bf.rxer.RemoveListener("nmp-rsp", bl)
for {
select {
- case <-bl.ErrChan:
- if err != nil {
+ case err, ok := <-bl.ErrChan:
+ if ok {
bf.errFunnel.Insert(err)
}
return
@@ -308,7 +302,7 @@ func (bf *BleFsm) connect() error {
}
defer bf.params.Bx.ReleaseMaster()
- bl, err := bf.rxer.AddSeqListener("connect", r.Seq)
+ bl, err := bf.rxer.AddListener("connect", SeqKey(r.Seq))
if err != nil {
return err
}
@@ -323,7 +317,7 @@ func (bf *BleFsm) connect() error {
bhe := nmxutil.ToBleHost(err)
if bhe != nil && bhe.Status == ERR_CODE_EDONE {
// Already connected.
- bf.rxer.RemoveSeqListener("connect", r.Seq)
+ bf.rxer.RemoveListener("connect", bl)
return nil
} else if !nmxutil.IsXport(err) {
// The transport did not restart; always attempt to cancel the
@@ -336,7 +330,7 @@ func (bf *BleFsm) connect() error {
}
}
- bf.rxer.RemoveSeqListener("connect", r.Seq)
+ bf.rxer.RemoveListener("connect", bl)
return err
}
@@ -378,11 +372,11 @@ func (bf *BleFsm) terminate() error {
r.ConnHandle = bf.connHandle
r.HciReason = ERR_CODE_HCI_REM_USER_CONN_TERM
- bl, err := bf.rxer.AddSeqListener("terminate", r.Seq)
+ bl, err := bf.rxer.AddListener("terminate", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("terminate", r.Seq)
+ defer bf.rxer.RemoveListener("terminate", bl)
if err := terminate(bf.params.Bx, bl, r); err != nil {
return err
@@ -394,11 +388,11 @@ func (bf *BleFsm) terminate() error {
func (bf *BleFsm) connCancel() error {
r := NewBleConnCancelReq()
- bl, err := bf.rxer.AddSeqListener("conn-cancel", r.Seq)
+ bl, err := bf.rxer.AddListener("conn-cancel", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("conn-cancel", r.Seq)
+ defer bf.rxer.RemoveListener("conn-cancel", bl)
if err := connCancel(bf.params.Bx, bl, r); err != nil {
return err
@@ -412,11 +406,11 @@ func (bf *BleFsm) discSvcUuidOnce(uuid BleUuid) (*BleSvc, error) {
r.ConnHandle = bf.connHandle
r.Uuid = uuid
- bl, err := bf.rxer.AddSeqListener("disc-svc-uuid", r.Seq)
+ bl, err := bf.rxer.AddListener("disc-svc-uuid", SeqKey(r.Seq))
if err != nil {
return nil, err
}
- defer bf.rxer.RemoveSeqListener("disc-svc-uuid", r.Seq)
+ defer bf.rxer.RemoveListener("disc-svc-uuid", bl)
svc, err := discSvcUuid(bf.params.Bx, bl, r)
if err != nil {
@@ -457,11 +451,11 @@ func (bf *BleFsm) encInitiate() error {
r := NewBleSecurityInitiateReq()
r.ConnHandle = bf.connHandle
- bl, err := bf.rxer.AddSeqListener("enc-initiate", r.Seq)
+ bl, err := bf.rxer.AddListener("enc-initiate", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("enc-initiate", r.Seq)
+ defer bf.rxer.RemoveListener("enc-initiate", bl)
// Initiate the encryption procedure.
if err := encInitiate(bf.params.Bx, bl, r); err != nil {
@@ -479,11 +473,11 @@ func (bf *BleFsm) discAllChrs() error {
r.StartHandle = bf.nmpSvc.StartHandle
r.EndHandle = bf.nmpSvc.EndHandle
- bl, err := bf.rxer.AddSeqListener("disc-all-chrs", r.Seq)
+ bl, err := bf.rxer.AddListener("disc-all-chrs", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("disc-all-chrs", r.Seq)
+ defer bf.rxer.RemoveListener("disc-all-chrs", bl)
chrs, err := discAllChrs(bf.params.Bx, bl, r)
if err != nil {
@@ -518,11 +512,11 @@ func (bf *BleFsm) exchangeMtu() error {
r := NewBleExchangeMtuReq()
r.ConnHandle = bf.connHandle
- bl, err := bf.rxer.AddSeqListener("exchange-mtu", r.Seq)
+ bl, err := bf.rxer.AddListener("exchange-mtu", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("exchange-mtu", r.Seq)
+ defer bf.rxer.RemoveListener("exchange-mtu", bl)
mtu, err := exchangeMtu(bf.params.Bx, bl, r)
if err != nil {
@@ -539,11 +533,11 @@ func (bf *BleFsm) writeCmd(data []byte) error {
r.AttrHandle = bf.nmpReqChr.ValHandle
r.Data.Bytes = data
- bl, err := bf.rxer.AddSeqListener("write-cmd", r.Seq)
+ bl, err := bf.rxer.AddListener("write-cmd", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("write-cmd", r.Seq)
+ defer bf.rxer.RemoveListener("write-cmd", bl)
if err := writeCmd(bf.params.Bx, bl, r); err != nil {
return err
@@ -558,11 +552,11 @@ func (bf *BleFsm) subscribe() error {
r.AttrHandle = bf.nmpRspChr.ValHandle + 1
r.Data.Bytes = []byte{1, 0}
- bl, err := bf.rxer.AddSeqListener("subscribe", r.Seq)
+ bl, err := bf.rxer.AddListener("subscribe", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("subscribe", r.Seq)
+ defer bf.rxer.RemoveListener("subscribe", bl)
if err := writeCmd(bf.params.Bx, bl, r); err != nil {
return err
diff --git a/nmxact/nmble/ble_util.go b/nmxact/nmble/ble_util.go
index 3550e33..79f2fe3 100644
--- a/nmxact/nmble/ble_util.go
+++ b/nmxact/nmble/ble_util.go
@@ -297,36 +297,25 @@ func ConnFindXact(x *BleXport, connHandle uint16) (BleConnDesc, error) {
r := NewBleConnFindReq()
r.ConnHandle = connHandle
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return BleConnDesc{}, err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return connFind(x, bl, r)
}
func GenRandAddrXact(x *BleXport) (BleAddr, error) {
r := NewBleGenRandAddrReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return BleAddr{}, err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return genRandAddr(x, bl, r)
}
@@ -335,18 +324,12 @@ func SetRandAddrXact(x *BleXport, addr BleAddr) error {
r := NewBleSetRandAddrReq()
r.Addr = addr
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return setRandAddr(x, bl, r)
}
@@ -355,18 +338,12 @@ func SetPreferredMtuXact(x *BleXport, mtu uint16) error {
r := NewBleSetPreferredMtuReq()
r.Mtu = mtu
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return setPreferredMtu(x, bl, r)
}
@@ -374,18 +351,12 @@ func SetPreferredMtuXact(x *BleXport, mtu uint16) error {
func ResetXact(x *BleXport) error {
r := NewResetReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return reset(x, bl, r)
}
diff --git a/nmxact/nmble/ble_xport.go b/nmxact/nmble/ble_xport.go
index 188a78f..7f1e210 100644
--- a/nmxact/nmble/ble_xport.go
+++ b/nmxact/nmble/ble_xport.go
@@ -163,28 +163,7 @@ func (bx *BleXport) BuildSesn(cfg sesn.SesnCfg) (sesn.Sesn, error) {
}
func (bx *BleXport) addSyncListener() (*Listener, error) {
- bl := NewListener()
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_SYNC_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: -1,
- }
- if err := bx.d.AddListener(base, bl); err != nil {
- return nil, err
- }
-
- return bl, nil
-}
-
-func (bx *BleXport) removeSyncListener() {
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_SYNC_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: -1,
- }
- bx.d.RemoveListener(base)
+ return bx.AddListener(TchKey(MSG_TYPE_SYNC_EVT, -1))
}
func (bx *BleXport) querySyncStatus() (bool, error) {
@@ -199,17 +178,12 @@ func (bx *BleXport) querySyncStatus() (bool, error) {
return false, err
}
- bl := NewListener()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: req.Seq,
- ConnHandle: -1,
- }
- if err := bx.d.AddListener(base, bl); err != nil {
+ key := SeqKey(req.Seq)
+ bl, err := bx.AddListener(key)
+ if err != nil {
return false, err
}
- defer bx.d.RemoveListener(base)
+ defer bx.RemoveListener(bl)
if err := bx.txNoSync(j); err != nil {
return false, err
@@ -235,7 +209,7 @@ func (bx *BleXport) initialSyncCheck() (bool, *Listener, error) {
synced, err := bx.querySyncStatus()
if err != nil {
- bx.removeSyncListener()
+ bx.RemoveListener(bl)
return false, nil, err
}
@@ -541,12 +515,20 @@ func (bx *BleXport) Tx(data []byte) error {
return bx.txNoSync(data)
}
-func (bx *BleXport) AddListener(base MsgBase, listener *Listener) error {
- return bx.d.AddListener(base, listener)
+func (bx *BleXport) AddListener(key ListenerKey) (*Listener, error) {
+ listener := NewListener()
+ if err := bx.d.AddListener(key, listener); err != nil {
+ return nil, err
+ }
+ return listener, nil
+}
+
+func (bx *BleXport) RemoveListener(listener *Listener) *ListenerKey {
+ return bx.d.RemoveListener(listener)
}
-func (bx *BleXport) RemoveListener(base MsgBase) *Listener {
- return bx.d.RemoveListener(base)
+func (bx *BleXport) RemoveKey(key ListenerKey) *Listener {
+ return bx.d.RemoveKey(key)
}
func (bx *BleXport) RspTimeout() time.Duration {
diff --git a/nmxact/nmble/discover.go b/nmxact/nmble/discover.go
index c8d4a3f..2e5f96a 100644
--- a/nmxact/nmble/discover.go
+++ b/nmxact/nmble/discover.go
@@ -32,18 +32,12 @@ func NewDiscoverer(params DiscovererParams) *Discoverer {
func (d *Discoverer) scanCancel() error {
r := NewBleScanCancelReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := d.params.Bx.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := d.params.Bx.AddListener(key)
+ if err != nil {
return err
}
- defer d.params.Bx.RemoveListener(base)
+ defer d.params.Bx.RemoveListener(bl)
if err := scanCancel(d.params.Bx, bl, r); err != nil {
return err
@@ -71,23 +65,17 @@ func (d *Discoverer) Start(advRptCb BleAdvRptFn) error {
r.Passive = d.params.Passive
r.FilterDuplicates = true
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := d.params.Bx.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := d.params.Bx.AddListener(key)
+ if err != nil {
return err
}
- defer d.params.Bx.RemoveListener(base)
+ defer d.params.Bx.RemoveListener(bl)
d.abortChan = make(chan struct{}, 1)
defer func() { d.abortChan = nil }()
- err := actScan(d.params.Bx, bl, r, d.abortChan, advRptCb)
+ err = actScan(d.params.Bx, bl, r, d.abortChan, advRptCb)
if !nmxutil.IsXport(err) {
// The transport did not restart; always attempt to cancel the scan
// operation. In some cases, the host has already stopped scanning
diff --git a/nmxact/nmble/dispatch.go b/nmxact/nmble/dispatch.go
index 76bd18c..45b76ac 100644
--- a/nmxact/nmble/dispatch.go
+++ b/nmxact/nmble/dispatch.go
@@ -23,16 +23,12 @@ import (
"encoding/json"
"fmt"
"sync"
- "time"
+
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
log "github.com/Sirupsen/logrus"
)
-type OpTypePair struct {
- Op MsgOp
- Type MsgType
-}
-
type MsgBase struct {
// Header
Op MsgOp `json:"op"`
@@ -43,49 +39,16 @@ type MsgBase struct {
ConnHandle int `json:"conn_handle" json:",omitempty"`
}
-type Listener struct {
- MsgChan chan Msg
- ErrChan chan error
- TmoChan chan time.Time
- Acked bool
-
- timer *time.Timer
-}
-
-func NewListener() *Listener {
- return &Listener{
- MsgChan: make(chan Msg, 16),
- ErrChan: make(chan error, 1),
- TmoChan: make(chan time.Time, 1),
- }
-}
-
-func (bl *Listener) AfterTimeout(tmo time.Duration) <-chan time.Time {
- fn := func() {
- if !bl.Acked {
- bl.TmoChan <- time.Now()
- }
- }
- bl.timer = time.AfterFunc(tmo, fn)
- return bl.TmoChan
-}
-
-func (bl *Listener) Close() {
- if bl.timer != nil {
- bl.timer.Stop()
- }
-
- close(bl.MsgChan)
- close(bl.ErrChan)
- close(bl.TmoChan)
+type OpTypePair struct {
+ Op MsgOp
+ Type MsgType
}
// The dispatcher is the owner of the listeners it points to. Only the
// dispatcher writes to these listeners.
type Dispatcher struct {
- seqMap map[BleSeq]*Listener
- baseMap map[MsgBase]*Listener
- mtx sync.Mutex
+ lm *ListenerMap
+ mtx sync.Mutex
}
type msgCtor func() Msg
@@ -154,101 +117,54 @@ var msgCtorMap = map[OpTypePair]msgCtor{
func NewDispatcher() *Dispatcher {
return &Dispatcher{
- seqMap: map[BleSeq]*Listener{},
- baseMap: map[MsgBase]*Listener{},
+ lm: NewListenerMap(),
}
}
-func (d *Dispatcher) findBaseListener(base MsgBase) (
- MsgBase, *Listener) {
-
- for k, v := range d.baseMap {
- if k.Op != -1 && base.Op != -1 && k.Op != base.Op {
- continue
- }
- if k.Type != -1 && base.Type != -1 && k.Type != base.Type {
- continue
- }
- if k.ConnHandle != -1 && base.ConnHandle != -1 &&
- k.ConnHandle != base.ConnHandle {
-
- continue
- }
-
- return k, v
- }
-
- return base, nil
-}
-
-func (d *Dispatcher) findDupListener(base MsgBase) (
- MsgBase, *Listener) {
-
- if base.Seq != BLE_SEQ_NONE {
- return base, d.seqMap[base.Seq]
- }
+func (d *Dispatcher) AddListener(key ListenerKey, listener *Listener) error {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
- return d.findBaseListener(base)
+ return d.lm.AddListener(key, listener)
}
-func (d *Dispatcher) findListener(base MsgBase) (
- MsgBase, *Listener) {
+func (d *Dispatcher) RemoveListener(listener *Listener) *ListenerKey {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
- if base.Seq != BLE_SEQ_NONE {
- if bl := d.seqMap[base.Seq]; bl != nil {
- return base, bl
- }
+ key := d.lm.RemoveListener(listener)
+ if key == nil {
+ return nil
}
- return d.findBaseListener(base)
+ listener.Close()
+ return key
}
-func (d *Dispatcher) AddListener(base MsgBase,
- listener *Listener) error {
-
+func (d *Dispatcher) RemoveKey(key ListenerKey) *Listener {
d.mtx.Lock()
defer d.mtx.Unlock()
- if ob, old := d.findDupListener(base); old != nil {
- return fmt.Errorf(
- "Duplicate BLE listener;\n"+
- " old=op=%d type=%d seq=%d connHandle=%d\n"+
- " new=op=%d type=%d seq=%d connHandle=%d",
- ob.Op, ob.Type, ob.Seq, ob.ConnHandle,
- base.Op, base.Type, base.Seq, base.ConnHandle)
- }
-
- if base.Seq != BLE_SEQ_NONE {
- if base.Op != -1 ||
- base.Type != -1 ||
- base.ConnHandle != -1 {
- return fmt.Errorf(
- "Invalid listener base; non-wild seq with wild fields")
- }
-
- d.seqMap[base.Seq] = listener
- } else {
- d.baseMap[base] = listener
+ listener := d.lm.RemoveKey(key)
+ if listener == nil {
+ return nil
}
- return nil
+ listener.Close()
+ return listener
}
-func (d *Dispatcher) RemoveListener(base MsgBase) *Listener {
+func (d *Dispatcher) ErrorAll(err error) {
+ nmxutil.Assert(err != nil)
+
d.mtx.Lock()
- defer d.mtx.Unlock()
+ listeners := d.lm.ExtractAll()
+ d.mtx.Unlock()
- base, bl := d.findListener(base)
- if bl != nil {
- bl.Close()
- if base.Seq != BLE_SEQ_NONE {
- delete(d.seqMap, base.Seq)
- } else {
- delete(d.baseMap, base)
- }
+ for _, listener := range listeners {
+ listener.ErrChan <- err
+ listener.Close()
}
-
- return bl
}
func decodeBleBase(data []byte) (MsgBase, error) {
@@ -290,8 +206,9 @@ func (d *Dispatcher) Dispatch(data []byte) {
}
d.mtx.Lock()
- _, listener := d.findListener(base)
- d.mtx.Unlock()
+ defer d.mtx.Unlock()
+
+ _, listener := d.lm.FindListener(base.Seq, base.Type, base.ConnHandle)
if listener == nil {
log.Debugf(
@@ -302,28 +219,3 @@ func (d *Dispatcher) Dispatch(data []byte) {
listener.MsgChan <- msg
}
-
-func (d *Dispatcher) ErrorAll(err error) {
- if err == nil {
- panic("NIL ERROR")
- }
-
- d.mtx.Lock()
-
- m1 := d.seqMap
- d.seqMap = map[BleSeq]*Listener{}
-
- m2 := d.baseMap
- d.baseMap = map[MsgBase]*Listener{}
-
- d.mtx.Unlock()
-
- for _, bl := range m1 {
- bl.ErrChan <- err
- bl.Close()
- }
- for _, bl := range m2 {
- bl.ErrChan <- err
- bl.Close()
- }
-}
diff --git a/nmxact/nmble/listen.go b/nmxact/nmble/listen.go
new file mode 100644
index 0000000..3181948
--- /dev/null
+++ b/nmxact/nmble/listen.go
@@ -0,0 +1,179 @@
+/**
+ * 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 nmble
+
+import (
+ "fmt"
+ "time"
+
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
+)
+
+type ListenerKey struct {
+ // seq-key only.
+ Seq BleSeq
+
+ // tch-key only (type-conn-handle).
+ Type MsgType
+ ConnHandle int
+}
+
+func SeqKey(seq BleSeq) ListenerKey {
+ return ListenerKey{
+ Seq: seq,
+ Type: -1,
+ ConnHandle: -1,
+ }
+}
+
+func TchKey(typ MsgType, connHandle int) ListenerKey {
+ return ListenerKey{
+ Seq: BLE_SEQ_NONE,
+ Type: typ,
+ ConnHandle: connHandle,
+ }
+}
+
+type Listener struct {
+ MsgChan chan Msg
+ ErrChan chan error
+ TmoChan chan time.Time
+ Acked bool
+
+ timer *time.Timer
+}
+
+func NewListener() *Listener {
+ return &Listener{
+ MsgChan: make(chan Msg, 16),
+ ErrChan: make(chan error, 1),
+ TmoChan: make(chan time.Time, 1),
+ }
+}
+
+func (bl *Listener) AfterTimeout(tmo time.Duration) <-chan time.Time {
+ fn := func() {
+ if !bl.Acked {
+ bl.TmoChan <- time.Now()
+ }
+ }
+ bl.timer = time.AfterFunc(tmo, fn)
+ return bl.TmoChan
+}
+
+func (bl *Listener) Close() {
+ if bl.timer != nil {
+ bl.timer.Stop()
+ }
+
+ close(bl.MsgChan)
+ close(bl.ErrChan)
+ close(bl.TmoChan)
+}
+
+// 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(seq BleSeq, typ MsgType, connHandle int) (
+ ListenerKey, *Listener) {
+
+ var key ListenerKey
+
+ // First, find by sequence number.
+ key = SeqKey(seq)
+ if listener := lm.k2l[key]; listener != nil {
+ return key, listener
+ }
+
+ // Otherwise, find by other fields.
+ key = TchKey(typ, connHandle)
+ 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 BLE listener: %#v", key)
+ }
+
+ if _, ok := lm.l2k[listener]; ok {
+ nmxutil.Assert(false)
+ return fmt.Errorf("Duplicate BLE 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/nmble/receiver.go b/nmxact/nmble/receiver.go
index 231ad5e..9fe6d3c 100644
--- a/nmxact/nmble/receiver.go
+++ b/nmxact/nmble/receiver.go
@@ -13,8 +13,8 @@ import (
type Receiver struct {
id uint32
bx *BleXport
+ lm *ListenerMap
logDepth int
- bls map[*Listener]MsgBase
mtx sync.Mutex
wg sync.WaitGroup
}
@@ -24,87 +24,69 @@ func NewReceiver(id uint32, bx *BleXport, logDepth int) *Receiver {
id: id,
bx: bx,
logDepth: logDepth + 3,
- bls: map[*Listener]MsgBase{},
+ lm: NewListenerMap(),
}
}
-func (r *Receiver) addListener(name string, base MsgBase) (
+func (r *Receiver) AddListener(name string, key ListenerKey) (
*Listener, error) {
- nmxutil.LogAddListener(r.logDepth, base, r.id, name)
-
- bl := NewListener()
+ nmxutil.LogAddListener(r.logDepth, key, r.id, name)
r.mtx.Lock()
defer r.mtx.Unlock()
- if err := r.bx.AddListener(base, bl); err != nil {
+ bl, err := r.bx.AddListener(key)
+ if err != nil {
return nil, err
}
- r.bls[bl] = base
+ r.lm.AddListener(key, bl)
r.wg.Add(1)
return bl, nil
}
-func (r *Receiver) AddBaseListener(name string, base MsgBase) (
- *Listener, error) {
-
- return r.addListener(name, base)
-}
-
-func (r *Receiver) AddSeqListener(name string, seq BleSeq) (
- *Listener, error) {
-
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: seq,
- ConnHandle: -1,
- }
- return r.addListener(name, base)
-}
-
-func (r *Receiver) removeListener(name string, base MsgBase) *Listener {
- nmxutil.LogRemoveListener(r.logDepth, base, r.id, name)
-
+func (r *Receiver) RemoveKey(name string, key ListenerKey) *Listener {
r.mtx.Lock()
defer r.mtx.Unlock()
- bl := r.bx.RemoveListener(base)
- delete(r.bls, bl)
-
- if bl != nil {
- r.wg.Done()
+ r.bx.RemoveKey(key)
+ bl := r.lm.RemoveKey(key)
+ if bl == nil {
+ return nil
}
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ r.wg.Done()
return bl
}
-func (r *Receiver) RemoveBaseListener(name string, base MsgBase) {
- r.removeListener(name, base)
-}
+func (r *Receiver) RemoveListener(name string, listener *Listener) *ListenerKey {
+ r.mtx.Lock()
+ defer r.mtx.Unlock()
-func (r *Receiver) RemoveSeqListener(name string, seq BleSeq) {
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: seq,
- ConnHandle: -1,
+ r.bx.RemoveListener(listener)
+ key := r.lm.RemoveListener(listener)
+ if key == nil {
+ return nil
}
- r.removeListener(name, base)
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ r.wg.Done()
+ return key
}
func (r *Receiver) RemoveAll(name string) {
r.mtx.Lock()
- bls := r.bls
- r.bls = map[*Listener]MsgBase{}
+ bls := r.lm.ExtractAll()
r.mtx.Unlock()
- for _, base := range bls {
- r.removeListener(name, base)
+ for _, bl := range bls {
+ if key := r.bx.RemoveListener(bl); key != nil {
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ }
+ r.wg.Done()
}
}
diff --git a/nmxact/nmp/dispatch.go b/nmxact/nmp/dispatch.go
index 9051b5b..5f36b0b 100644
--- a/nmxact/nmp/dispatch.go
+++ b/nmxact/nmp/dispatch.go
@@ -127,6 +127,9 @@ func decodeRsp(pkt []byte) (NmpRsp, error) {
// Returns true if the response was dispatched.
func (d *Dispatcher) DispatchRsp(r NmpRsp) bool {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
+
log.Debugf("Received nmp rsp: %+v", r)
nl := d.seqListenerMap[r.Hdr().Seq]
@@ -142,9 +145,6 @@ func (d *Dispatcher) DispatchRsp(r NmpRsp) bool {
// Returns true if the response was dispatched.
func (d *Dispatcher) Dispatch(data []byte) bool {
- d.mtx.Lock()
- defer d.mtx.Unlock()
-
pkt := d.reassembler.RxFrag(data)
if pkt == nil {
return false
diff --git a/nmxact/nmxutil/nmxutil.go b/nmxact/nmxutil/nmxutil.go
index 078a06d..84f45ae 100644
--- a/nmxact/nmxutil/nmxutil.go
+++ b/nmxact/nmxutil/nmxutil.go
@@ -10,9 +10,10 @@ import (
"sync"
"time"
+ log "github.com/Sirupsen/logrus"
"github.com/ugorji/go/codec"
- log "github.com/Sirupsen/logrus"
+ "mynewt.apache.org/newt/util"
)
const DURATION_FOREVER time.Duration = math.MaxInt64
@@ -45,6 +46,7 @@ func SetLogLevel(level log.Level) {
func Assert(cond bool) {
if Debug && !cond {
+ util.PrintStacks()
panic("Failed assertion")
}
}
@@ -115,16 +117,16 @@ func LogRemoveOicListener(parentLevel int, token []byte) {
fmt.Sprintf("token=%+v", token))
}
-func LogAddListener(parentLevel int, base interface{}, id uint32,
+func LogAddListener(parentLevel int, key interface{}, id uint32,
name string) {
LogListener(parentLevel, "add-ble-listener",
- fmt.Sprintf("[%d] %s: base=%+v", id, name, base))
+ fmt.Sprintf("[%d] %s: base=%+v", id, name, key))
}
-func LogRemoveListener(parentLevel int, base interface{}, id uint32,
+func LogRemoveListener(parentLevel int, key interface{}, id uint32,
name string) {
LogListener(parentLevel, "remove-ble-listener",
- fmt.Sprintf("[%d] %s: base=%+v", id, name, base))
+ fmt.Sprintf("[%d] %s: base=%+v", id, name, key))
}
diff --git a/nmxact/oic/dispatch.go b/nmxact/oic/dispatch.go
index 810abe7..33f0ac9 100644
--- a/nmxact/oic/dispatch.go
+++ b/nmxact/oic/dispatch.go
@@ -125,25 +125,42 @@ func (d *Dispatcher) AddListener(token []byte) (*Listener, error) {
}
func (d *Dispatcher) RemoveListener(token []byte) *Listener {
- nmxutil.LogRemoveOicListener(d.logDepth, token)
-
d.mtx.Lock()
defer d.mtx.Unlock()
ot, err := NewToken(token)
if err != nil {
+ log.Errorf("Error creating OIC token: %s", err.Error())
return nil
}
ol := d.tokenListenerMap[ot]
- if ol != nil {
- ol.Close()
- delete(d.tokenListenerMap, ot)
+ if ol == nil {
+ return nil
}
+ nmxutil.LogRemoveOicListener(d.logDepth, token)
+ ol.Close()
+ delete(d.tokenListenerMap, ot)
+
return ol
}
+func (d *Dispatcher) dispatchRsp(ot Token, msg *coap.Message) bool {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
+
+ ol := d.tokenListenerMap[ot]
+
+ if ol == nil {
+ log.Printf("No listener for incoming OIC message; token=%#v", ot)
+ return false
+ }
+
+ ol.RspChan <- msg
+ return true
+}
+
// Returns true if the response was dispatched.
func (d *Dispatcher) Dispatch(data []byte) bool {
var msg *coap.Message
@@ -171,17 +188,7 @@ func (d *Dispatcher) Dispatch(data []byte) bool {
return false
}
- d.mtx.Lock()
- ol := d.tokenListenerMap[ot]
- d.mtx.Unlock()
-
- if ol == nil {
- log.Printf("No listener for incoming OIC message; token=%#v", ot)
- return false
- }
-
- ol.RspChan <- msg
- return true
+ return d.dispatchRsp(ot, msg)
}
func (d *Dispatcher) ErrorOne(token Token, err error) error {
diff --git a/nmxact/omp/dispatch.go b/nmxact/omp/dispatch.go
index 7647048..093efa5 100644
--- a/nmxact/omp/dispatch.go
+++ b/nmxact/omp/dispatch.go
@@ -38,33 +38,33 @@ type Dispatcher struct {
}
func NewDispatcher(isTcp bool, logDepth int) (*Dispatcher, error) {
- r := &Dispatcher{
+ d := &Dispatcher{
nmpd: nmp.NewDispatcher(logDepth + 1),
oicd: oic.NewDispatcher(isTcp, logDepth+1),
stopCh: make(chan struct{}),
}
// Listen for OMP responses. This should never fail.
- if err := r.addOmpListener(); err != nil {
+ if err := d.addOmpListener(); err != nil {
log.Errorf("Unexpected failure to add OMP listener: " + err.Error())
return nil, err
}
- return r, nil
+ return d, nil
}
-func (r *Dispatcher) addOmpListener() error {
+func (d *Dispatcher) addOmpListener() error {
// OMP responses are identifiable by the lack of a CoAP token. Set up a
// permanent listener to receive these messages.
- ol, err := r.AddOicListener(nil)
+ ol, err := d.AddOicListener(nil)
if err != nil {
return err
}
- r.wg.Add(1)
+ d.wg.Add(1)
go func() {
- defer r.RemoveOicListener(nil)
- defer r.wg.Done()
+ defer d.RemoveOicListener(nil)
+ defer d.wg.Done()
for {
select {
@@ -73,13 +73,13 @@ func (r *Dispatcher) addOmpListener() error {
if err != nil {
log.Debugf("OMP decode failure: %s", err.Error())
} else {
- r.nmpd.DispatchRsp(rsp)
+ d.nmpd.DispatchRsp(rsp)
}
case err := <-ol.ErrChan:
log.Debugf("OIC error: %s", err.Error())
- case <-r.stopCh:
+ case <-d.stopCh:
return
}
}
@@ -88,36 +88,36 @@ func (r *Dispatcher) addOmpListener() error {
return nil
}
-func (r *Dispatcher) Stop() {
- r.stopCh <- struct{}{}
- r.wg.Wait()
+func (d *Dispatcher) Stop() {
+ d.stopCh <- struct{}{}
+ d.wg.Wait()
}
-func (r *Dispatcher) Dispatch(data []byte) bool {
- return r.oicd.Dispatch(data)
+func (d *Dispatcher) Dispatch(data []byte) bool {
+ return d.oicd.Dispatch(data)
}
-func (r *Dispatcher) AddOicListener(token []byte) (*oic.Listener, error) {
- return r.oicd.AddListener(token)
+func (d *Dispatcher) AddOicListener(token []byte) (*oic.Listener, error) {
+ return d.oicd.AddListener(token)
}
-func (r *Dispatcher) RemoveOicListener(token []byte) *oic.Listener {
- return r.oicd.RemoveListener(token)
+func (d *Dispatcher) RemoveOicListener(token []byte) *oic.Listener {
+ return d.oicd.RemoveListener(token)
}
-func (r *Dispatcher) AddNmpListener(seq uint8) (*nmp.Listener, error) {
- return r.nmpd.AddListener(seq)
+func (d *Dispatcher) AddNmpListener(seq uint8) (*nmp.Listener, error) {
+ return d.nmpd.AddListener(seq)
}
-func (r *Dispatcher) RemoveNmpListener(seq uint8) *nmp.Listener {
- return r.nmpd.RemoveListener(seq)
+func (d *Dispatcher) RemoveNmpListener(seq uint8) *nmp.Listener {
+ return d.nmpd.RemoveListener(seq)
}
-func (r *Dispatcher) ErrorOneNmp(seq uint8, err error) error {
- return r.nmpd.ErrorOne(seq, err)
+func (d *Dispatcher) ErrorOneNmp(seq uint8, err error) error {
+ return d.nmpd.ErrorOne(seq, err)
}
-func (r *Dispatcher) ErrorAll(err error) {
- r.nmpd.ErrorAll(err)
- r.oicd.ErrorAll(err)
+func (d *Dispatcher) ErrorAll(err error) {
+ d.nmpd.ErrorAll(err)
+ d.oicd.ErrorAll(err)
}
--
To stop receiving notification emails like this one, please contact
"commits@mynewt.apache.org" <co...@mynewt.apache.org>.
[mynewt-newtmgr] 02/02: newtmgr - revendor
Posted by cc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ccollins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git
commit 86d517dc4b764e0496f0fee0baef08850c4f4d3f
Author: Christopher Collins <cc...@apache.org>
AuthorDate: Sun Jul 16 13:26:08 2017 -0700
newtmgr - revendor
---
newtmgr/Godeps/Godeps.json | 48 +++---
.../newtmgr/nmxact/nmble/ble_fsm.go | 56 +++----
.../newtmgr/nmxact/nmble/ble_util.go | 69 +++-----
.../newtmgr/nmxact/nmble/ble_xport.go | 54 ++----
.../newtmgr/nmxact/nmble/discover.go | 30 +---
.../newtmgr/nmxact/nmble/dispatch.go | 182 +++++----------------
.../newtmgr/nmxact/nmble/listen.go | 179 ++++++++++++++++++++
.../newtmgr/nmxact/nmble/receiver.go | 78 ++++-----
.../newtmgr/nmxact/nmp/dispatch.go | 6 +-
.../newtmgr/nmxact/nmxutil/nmxutil.go | 12 +-
.../newtmgr/nmxact/oic/dispatch.go | 39 +++--
.../newtmgr/nmxact/omp/dispatch.go | 56 +++----
12 files changed, 403 insertions(+), 406 deletions(-)
diff --git a/newtmgr/Godeps/Godeps.json b/newtmgr/Godeps/Godeps.json
index d205eef..8e386a3 100644
--- a/newtmgr/Godeps/Godeps.json
+++ b/newtmgr/Godeps/Godeps.json
@@ -126,63 +126,63 @@
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/bledefs",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/nmble",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/nmp",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/nmserial",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/nmxutil",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/oic",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/omp",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/scan",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/sesn",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/udp",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/xact",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
},
{
"ImportPath": "mynewt.apache.org/newtmgr/nmxact/xport",
- "Comment": "mynewt_0_9_0_tag-552-g62a3d9e",
- "Rev": "62a3d9ece4c5ad7b8b357d66ee8a61e4e40c4063"
+ "Comment": "mynewt_0_9_0_tag-554-gf64449a",
+ "Rev": "f64449a63215d75d533034dfc62dca4678c5c23c"
}
]
}
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_fsm.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_fsm.go
index 6721eb0..8d80bfd 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_fsm.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_fsm.go
@@ -203,7 +203,7 @@ func (bf *BleFsm) eventListen(bl *Listener, seq BleSeq) error {
go func() {
defer bf.wg.Done()
- defer bf.rxer.RemoveSeqListener("connect", seq)
+ defer bf.rxer.RemoveListener("connect", bl)
for {
select {
@@ -254,14 +254,8 @@ func (bf *BleFsm) eventListen(bl *Listener, seq BleSeq) error {
}
func (bf *BleFsm) nmpRspListen() error {
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_NOTIFY_RX_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: int(bf.connHandle),
- }
-
- bl, err := bf.rxer.AddBaseListener("nmp-rsp", base)
+ key := TchKey(MSG_TYPE_NOTIFY_RX_EVT, int(bf.connHandle))
+ bl, err := bf.rxer.AddListener("nmp-rsp", key)
if err != nil {
return err
}
@@ -270,12 +264,12 @@ func (bf *BleFsm) nmpRspListen() error {
go func() {
defer bf.wg.Done()
- defer bf.rxer.RemoveBaseListener("nmp-rsp", base)
+ defer bf.rxer.RemoveListener("nmp-rsp", bl)
for {
select {
- case <-bl.ErrChan:
- if err != nil {
+ case err, ok := <-bl.ErrChan:
+ if ok {
bf.errFunnel.Insert(err)
}
return
@@ -308,7 +302,7 @@ func (bf *BleFsm) connect() error {
}
defer bf.params.Bx.ReleaseMaster()
- bl, err := bf.rxer.AddSeqListener("connect", r.Seq)
+ bl, err := bf.rxer.AddListener("connect", SeqKey(r.Seq))
if err != nil {
return err
}
@@ -323,7 +317,7 @@ func (bf *BleFsm) connect() error {
bhe := nmxutil.ToBleHost(err)
if bhe != nil && bhe.Status == ERR_CODE_EDONE {
// Already connected.
- bf.rxer.RemoveSeqListener("connect", r.Seq)
+ bf.rxer.RemoveListener("connect", bl)
return nil
} else if !nmxutil.IsXport(err) {
// The transport did not restart; always attempt to cancel the
@@ -336,7 +330,7 @@ func (bf *BleFsm) connect() error {
}
}
- bf.rxer.RemoveSeqListener("connect", r.Seq)
+ bf.rxer.RemoveListener("connect", bl)
return err
}
@@ -378,11 +372,11 @@ func (bf *BleFsm) terminate() error {
r.ConnHandle = bf.connHandle
r.HciReason = ERR_CODE_HCI_REM_USER_CONN_TERM
- bl, err := bf.rxer.AddSeqListener("terminate", r.Seq)
+ bl, err := bf.rxer.AddListener("terminate", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("terminate", r.Seq)
+ defer bf.rxer.RemoveListener("terminate", bl)
if err := terminate(bf.params.Bx, bl, r); err != nil {
return err
@@ -394,11 +388,11 @@ func (bf *BleFsm) terminate() error {
func (bf *BleFsm) connCancel() error {
r := NewBleConnCancelReq()
- bl, err := bf.rxer.AddSeqListener("conn-cancel", r.Seq)
+ bl, err := bf.rxer.AddListener("conn-cancel", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("conn-cancel", r.Seq)
+ defer bf.rxer.RemoveListener("conn-cancel", bl)
if err := connCancel(bf.params.Bx, bl, r); err != nil {
return err
@@ -412,11 +406,11 @@ func (bf *BleFsm) discSvcUuidOnce(uuid BleUuid) (*BleSvc, error) {
r.ConnHandle = bf.connHandle
r.Uuid = uuid
- bl, err := bf.rxer.AddSeqListener("disc-svc-uuid", r.Seq)
+ bl, err := bf.rxer.AddListener("disc-svc-uuid", SeqKey(r.Seq))
if err != nil {
return nil, err
}
- defer bf.rxer.RemoveSeqListener("disc-svc-uuid", r.Seq)
+ defer bf.rxer.RemoveListener("disc-svc-uuid", bl)
svc, err := discSvcUuid(bf.params.Bx, bl, r)
if err != nil {
@@ -457,11 +451,11 @@ func (bf *BleFsm) encInitiate() error {
r := NewBleSecurityInitiateReq()
r.ConnHandle = bf.connHandle
- bl, err := bf.rxer.AddSeqListener("enc-initiate", r.Seq)
+ bl, err := bf.rxer.AddListener("enc-initiate", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("enc-initiate", r.Seq)
+ defer bf.rxer.RemoveListener("enc-initiate", bl)
// Initiate the encryption procedure.
if err := encInitiate(bf.params.Bx, bl, r); err != nil {
@@ -479,11 +473,11 @@ func (bf *BleFsm) discAllChrs() error {
r.StartHandle = bf.nmpSvc.StartHandle
r.EndHandle = bf.nmpSvc.EndHandle
- bl, err := bf.rxer.AddSeqListener("disc-all-chrs", r.Seq)
+ bl, err := bf.rxer.AddListener("disc-all-chrs", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("disc-all-chrs", r.Seq)
+ defer bf.rxer.RemoveListener("disc-all-chrs", bl)
chrs, err := discAllChrs(bf.params.Bx, bl, r)
if err != nil {
@@ -518,11 +512,11 @@ func (bf *BleFsm) exchangeMtu() error {
r := NewBleExchangeMtuReq()
r.ConnHandle = bf.connHandle
- bl, err := bf.rxer.AddSeqListener("exchange-mtu", r.Seq)
+ bl, err := bf.rxer.AddListener("exchange-mtu", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("exchange-mtu", r.Seq)
+ defer bf.rxer.RemoveListener("exchange-mtu", bl)
mtu, err := exchangeMtu(bf.params.Bx, bl, r)
if err != nil {
@@ -539,11 +533,11 @@ func (bf *BleFsm) writeCmd(data []byte) error {
r.AttrHandle = bf.nmpReqChr.ValHandle
r.Data.Bytes = data
- bl, err := bf.rxer.AddSeqListener("write-cmd", r.Seq)
+ bl, err := bf.rxer.AddListener("write-cmd", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("write-cmd", r.Seq)
+ defer bf.rxer.RemoveListener("write-cmd", bl)
if err := writeCmd(bf.params.Bx, bl, r); err != nil {
return err
@@ -558,11 +552,11 @@ func (bf *BleFsm) subscribe() error {
r.AttrHandle = bf.nmpRspChr.ValHandle + 1
r.Data.Bytes = []byte{1, 0}
- bl, err := bf.rxer.AddSeqListener("subscribe", r.Seq)
+ bl, err := bf.rxer.AddListener("subscribe", SeqKey(r.Seq))
if err != nil {
return err
}
- defer bf.rxer.RemoveSeqListener("subscribe", r.Seq)
+ defer bf.rxer.RemoveListener("subscribe", bl)
if err := writeCmd(bf.params.Bx, bl, r); err != nil {
return err
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_util.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_util.go
index 3550e33..79f2fe3 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_util.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_util.go
@@ -297,36 +297,25 @@ func ConnFindXact(x *BleXport, connHandle uint16) (BleConnDesc, error) {
r := NewBleConnFindReq()
r.ConnHandle = connHandle
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return BleConnDesc{}, err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return connFind(x, bl, r)
}
func GenRandAddrXact(x *BleXport) (BleAddr, error) {
r := NewBleGenRandAddrReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return BleAddr{}, err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return genRandAddr(x, bl, r)
}
@@ -335,18 +324,12 @@ func SetRandAddrXact(x *BleXport, addr BleAddr) error {
r := NewBleSetRandAddrReq()
r.Addr = addr
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return setRandAddr(x, bl, r)
}
@@ -355,18 +338,12 @@ func SetPreferredMtuXact(x *BleXport, mtu uint16) error {
r := NewBleSetPreferredMtuReq()
r.Mtu = mtu
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return setPreferredMtu(x, bl, r)
}
@@ -374,18 +351,12 @@ func SetPreferredMtuXact(x *BleXport, mtu uint16) error {
func ResetXact(x *BleXport) error {
r := NewResetReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := x.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := x.AddListener(key)
+ if err != nil {
return err
}
- defer x.RemoveListener(base)
+ defer x.RemoveListener(bl)
return reset(x, bl, r)
}
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_xport.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_xport.go
index 188a78f..7f1e210 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_xport.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/ble_xport.go
@@ -163,28 +163,7 @@ func (bx *BleXport) BuildSesn(cfg sesn.SesnCfg) (sesn.Sesn, error) {
}
func (bx *BleXport) addSyncListener() (*Listener, error) {
- bl := NewListener()
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_SYNC_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: -1,
- }
- if err := bx.d.AddListener(base, bl); err != nil {
- return nil, err
- }
-
- return bl, nil
-}
-
-func (bx *BleXport) removeSyncListener() {
- base := MsgBase{
- Op: MSG_OP_EVT,
- Type: MSG_TYPE_SYNC_EVT,
- Seq: BLE_SEQ_NONE,
- ConnHandle: -1,
- }
- bx.d.RemoveListener(base)
+ return bx.AddListener(TchKey(MSG_TYPE_SYNC_EVT, -1))
}
func (bx *BleXport) querySyncStatus() (bool, error) {
@@ -199,17 +178,12 @@ func (bx *BleXport) querySyncStatus() (bool, error) {
return false, err
}
- bl := NewListener()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: req.Seq,
- ConnHandle: -1,
- }
- if err := bx.d.AddListener(base, bl); err != nil {
+ key := SeqKey(req.Seq)
+ bl, err := bx.AddListener(key)
+ if err != nil {
return false, err
}
- defer bx.d.RemoveListener(base)
+ defer bx.RemoveListener(bl)
if err := bx.txNoSync(j); err != nil {
return false, err
@@ -235,7 +209,7 @@ func (bx *BleXport) initialSyncCheck() (bool, *Listener, error) {
synced, err := bx.querySyncStatus()
if err != nil {
- bx.removeSyncListener()
+ bx.RemoveListener(bl)
return false, nil, err
}
@@ -541,12 +515,20 @@ func (bx *BleXport) Tx(data []byte) error {
return bx.txNoSync(data)
}
-func (bx *BleXport) AddListener(base MsgBase, listener *Listener) error {
- return bx.d.AddListener(base, listener)
+func (bx *BleXport) AddListener(key ListenerKey) (*Listener, error) {
+ listener := NewListener()
+ if err := bx.d.AddListener(key, listener); err != nil {
+ return nil, err
+ }
+ return listener, nil
+}
+
+func (bx *BleXport) RemoveListener(listener *Listener) *ListenerKey {
+ return bx.d.RemoveListener(listener)
}
-func (bx *BleXport) RemoveListener(base MsgBase) *Listener {
- return bx.d.RemoveListener(base)
+func (bx *BleXport) RemoveKey(key ListenerKey) *Listener {
+ return bx.d.RemoveKey(key)
}
func (bx *BleXport) RspTimeout() time.Duration {
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/discover.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/discover.go
index c8d4a3f..2e5f96a 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/discover.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/discover.go
@@ -32,18 +32,12 @@ func NewDiscoverer(params DiscovererParams) *Discoverer {
func (d *Discoverer) scanCancel() error {
r := NewBleScanCancelReq()
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := d.params.Bx.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := d.params.Bx.AddListener(key)
+ if err != nil {
return err
}
- defer d.params.Bx.RemoveListener(base)
+ defer d.params.Bx.RemoveListener(bl)
if err := scanCancel(d.params.Bx, bl, r); err != nil {
return err
@@ -71,23 +65,17 @@ func (d *Discoverer) Start(advRptCb BleAdvRptFn) error {
r.Passive = d.params.Passive
r.FilterDuplicates = true
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: r.Seq,
- ConnHandle: -1,
- }
-
- bl := NewListener()
- if err := d.params.Bx.AddListener(base, bl); err != nil {
+ key := SeqKey(r.Seq)
+ bl, err := d.params.Bx.AddListener(key)
+ if err != nil {
return err
}
- defer d.params.Bx.RemoveListener(base)
+ defer d.params.Bx.RemoveListener(bl)
d.abortChan = make(chan struct{}, 1)
defer func() { d.abortChan = nil }()
- err := actScan(d.params.Bx, bl, r, d.abortChan, advRptCb)
+ err = actScan(d.params.Bx, bl, r, d.abortChan, advRptCb)
if !nmxutil.IsXport(err) {
// The transport did not restart; always attempt to cancel the scan
// operation. In some cases, the host has already stopped scanning
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/dispatch.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/dispatch.go
index 76bd18c..45b76ac 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/dispatch.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/dispatch.go
@@ -23,16 +23,12 @@ import (
"encoding/json"
"fmt"
"sync"
- "time"
+
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
log "github.com/Sirupsen/logrus"
)
-type OpTypePair struct {
- Op MsgOp
- Type MsgType
-}
-
type MsgBase struct {
// Header
Op MsgOp `json:"op"`
@@ -43,49 +39,16 @@ type MsgBase struct {
ConnHandle int `json:"conn_handle" json:",omitempty"`
}
-type Listener struct {
- MsgChan chan Msg
- ErrChan chan error
- TmoChan chan time.Time
- Acked bool
-
- timer *time.Timer
-}
-
-func NewListener() *Listener {
- return &Listener{
- MsgChan: make(chan Msg, 16),
- ErrChan: make(chan error, 1),
- TmoChan: make(chan time.Time, 1),
- }
-}
-
-func (bl *Listener) AfterTimeout(tmo time.Duration) <-chan time.Time {
- fn := func() {
- if !bl.Acked {
- bl.TmoChan <- time.Now()
- }
- }
- bl.timer = time.AfterFunc(tmo, fn)
- return bl.TmoChan
-}
-
-func (bl *Listener) Close() {
- if bl.timer != nil {
- bl.timer.Stop()
- }
-
- close(bl.MsgChan)
- close(bl.ErrChan)
- close(bl.TmoChan)
+type OpTypePair struct {
+ Op MsgOp
+ Type MsgType
}
// The dispatcher is the owner of the listeners it points to. Only the
// dispatcher writes to these listeners.
type Dispatcher struct {
- seqMap map[BleSeq]*Listener
- baseMap map[MsgBase]*Listener
- mtx sync.Mutex
+ lm *ListenerMap
+ mtx sync.Mutex
}
type msgCtor func() Msg
@@ -154,101 +117,54 @@ var msgCtorMap = map[OpTypePair]msgCtor{
func NewDispatcher() *Dispatcher {
return &Dispatcher{
- seqMap: map[BleSeq]*Listener{},
- baseMap: map[MsgBase]*Listener{},
+ lm: NewListenerMap(),
}
}
-func (d *Dispatcher) findBaseListener(base MsgBase) (
- MsgBase, *Listener) {
-
- for k, v := range d.baseMap {
- if k.Op != -1 && base.Op != -1 && k.Op != base.Op {
- continue
- }
- if k.Type != -1 && base.Type != -1 && k.Type != base.Type {
- continue
- }
- if k.ConnHandle != -1 && base.ConnHandle != -1 &&
- k.ConnHandle != base.ConnHandle {
-
- continue
- }
-
- return k, v
- }
-
- return base, nil
-}
-
-func (d *Dispatcher) findDupListener(base MsgBase) (
- MsgBase, *Listener) {
-
- if base.Seq != BLE_SEQ_NONE {
- return base, d.seqMap[base.Seq]
- }
+func (d *Dispatcher) AddListener(key ListenerKey, listener *Listener) error {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
- return d.findBaseListener(base)
+ return d.lm.AddListener(key, listener)
}
-func (d *Dispatcher) findListener(base MsgBase) (
- MsgBase, *Listener) {
+func (d *Dispatcher) RemoveListener(listener *Listener) *ListenerKey {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
- if base.Seq != BLE_SEQ_NONE {
- if bl := d.seqMap[base.Seq]; bl != nil {
- return base, bl
- }
+ key := d.lm.RemoveListener(listener)
+ if key == nil {
+ return nil
}
- return d.findBaseListener(base)
+ listener.Close()
+ return key
}
-func (d *Dispatcher) AddListener(base MsgBase,
- listener *Listener) error {
-
+func (d *Dispatcher) RemoveKey(key ListenerKey) *Listener {
d.mtx.Lock()
defer d.mtx.Unlock()
- if ob, old := d.findDupListener(base); old != nil {
- return fmt.Errorf(
- "Duplicate BLE listener;\n"+
- " old=op=%d type=%d seq=%d connHandle=%d\n"+
- " new=op=%d type=%d seq=%d connHandle=%d",
- ob.Op, ob.Type, ob.Seq, ob.ConnHandle,
- base.Op, base.Type, base.Seq, base.ConnHandle)
- }
-
- if base.Seq != BLE_SEQ_NONE {
- if base.Op != -1 ||
- base.Type != -1 ||
- base.ConnHandle != -1 {
- return fmt.Errorf(
- "Invalid listener base; non-wild seq with wild fields")
- }
-
- d.seqMap[base.Seq] = listener
- } else {
- d.baseMap[base] = listener
+ listener := d.lm.RemoveKey(key)
+ if listener == nil {
+ return nil
}
- return nil
+ listener.Close()
+ return listener
}
-func (d *Dispatcher) RemoveListener(base MsgBase) *Listener {
+func (d *Dispatcher) ErrorAll(err error) {
+ nmxutil.Assert(err != nil)
+
d.mtx.Lock()
- defer d.mtx.Unlock()
+ listeners := d.lm.ExtractAll()
+ d.mtx.Unlock()
- base, bl := d.findListener(base)
- if bl != nil {
- bl.Close()
- if base.Seq != BLE_SEQ_NONE {
- delete(d.seqMap, base.Seq)
- } else {
- delete(d.baseMap, base)
- }
+ for _, listener := range listeners {
+ listener.ErrChan <- err
+ listener.Close()
}
-
- return bl
}
func decodeBleBase(data []byte) (MsgBase, error) {
@@ -290,8 +206,9 @@ func (d *Dispatcher) Dispatch(data []byte) {
}
d.mtx.Lock()
- _, listener := d.findListener(base)
- d.mtx.Unlock()
+ defer d.mtx.Unlock()
+
+ _, listener := d.lm.FindListener(base.Seq, base.Type, base.ConnHandle)
if listener == nil {
log.Debugf(
@@ -302,28 +219,3 @@ func (d *Dispatcher) Dispatch(data []byte) {
listener.MsgChan <- msg
}
-
-func (d *Dispatcher) ErrorAll(err error) {
- if err == nil {
- panic("NIL ERROR")
- }
-
- d.mtx.Lock()
-
- m1 := d.seqMap
- d.seqMap = map[BleSeq]*Listener{}
-
- m2 := d.baseMap
- d.baseMap = map[MsgBase]*Listener{}
-
- d.mtx.Unlock()
-
- for _, bl := range m1 {
- bl.ErrChan <- err
- bl.Close()
- }
- for _, bl := range m2 {
- bl.ErrChan <- err
- bl.Close()
- }
-}
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/listen.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/listen.go
new file mode 100644
index 0000000..3181948
--- /dev/null
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/listen.go
@@ -0,0 +1,179 @@
+/**
+ * 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 nmble
+
+import (
+ "fmt"
+ "time"
+
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
+)
+
+type ListenerKey struct {
+ // seq-key only.
+ Seq BleSeq
+
+ // tch-key only (type-conn-handle).
+ Type MsgType
+ ConnHandle int
+}
+
+func SeqKey(seq BleSeq) ListenerKey {
+ return ListenerKey{
+ Seq: seq,
+ Type: -1,
+ ConnHandle: -1,
+ }
+}
+
+func TchKey(typ MsgType, connHandle int) ListenerKey {
+ return ListenerKey{
+ Seq: BLE_SEQ_NONE,
+ Type: typ,
+ ConnHandle: connHandle,
+ }
+}
+
+type Listener struct {
+ MsgChan chan Msg
+ ErrChan chan error
+ TmoChan chan time.Time
+ Acked bool
+
+ timer *time.Timer
+}
+
+func NewListener() *Listener {
+ return &Listener{
+ MsgChan: make(chan Msg, 16),
+ ErrChan: make(chan error, 1),
+ TmoChan: make(chan time.Time, 1),
+ }
+}
+
+func (bl *Listener) AfterTimeout(tmo time.Duration) <-chan time.Time {
+ fn := func() {
+ if !bl.Acked {
+ bl.TmoChan <- time.Now()
+ }
+ }
+ bl.timer = time.AfterFunc(tmo, fn)
+ return bl.TmoChan
+}
+
+func (bl *Listener) Close() {
+ if bl.timer != nil {
+ bl.timer.Stop()
+ }
+
+ close(bl.MsgChan)
+ close(bl.ErrChan)
+ close(bl.TmoChan)
+}
+
+// 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(seq BleSeq, typ MsgType, connHandle int) (
+ ListenerKey, *Listener) {
+
+ var key ListenerKey
+
+ // First, find by sequence number.
+ key = SeqKey(seq)
+ if listener := lm.k2l[key]; listener != nil {
+ return key, listener
+ }
+
+ // Otherwise, find by other fields.
+ key = TchKey(typ, connHandle)
+ 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 BLE listener: %#v", key)
+ }
+
+ if _, ok := lm.l2k[listener]; ok {
+ nmxutil.Assert(false)
+ return fmt.Errorf("Duplicate BLE 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/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/receiver.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/receiver.go
index 231ad5e..9fe6d3c 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/receiver.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmble/receiver.go
@@ -13,8 +13,8 @@ import (
type Receiver struct {
id uint32
bx *BleXport
+ lm *ListenerMap
logDepth int
- bls map[*Listener]MsgBase
mtx sync.Mutex
wg sync.WaitGroup
}
@@ -24,87 +24,69 @@ func NewReceiver(id uint32, bx *BleXport, logDepth int) *Receiver {
id: id,
bx: bx,
logDepth: logDepth + 3,
- bls: map[*Listener]MsgBase{},
+ lm: NewListenerMap(),
}
}
-func (r *Receiver) addListener(name string, base MsgBase) (
+func (r *Receiver) AddListener(name string, key ListenerKey) (
*Listener, error) {
- nmxutil.LogAddListener(r.logDepth, base, r.id, name)
-
- bl := NewListener()
+ nmxutil.LogAddListener(r.logDepth, key, r.id, name)
r.mtx.Lock()
defer r.mtx.Unlock()
- if err := r.bx.AddListener(base, bl); err != nil {
+ bl, err := r.bx.AddListener(key)
+ if err != nil {
return nil, err
}
- r.bls[bl] = base
+ r.lm.AddListener(key, bl)
r.wg.Add(1)
return bl, nil
}
-func (r *Receiver) AddBaseListener(name string, base MsgBase) (
- *Listener, error) {
-
- return r.addListener(name, base)
-}
-
-func (r *Receiver) AddSeqListener(name string, seq BleSeq) (
- *Listener, error) {
-
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: seq,
- ConnHandle: -1,
- }
- return r.addListener(name, base)
-}
-
-func (r *Receiver) removeListener(name string, base MsgBase) *Listener {
- nmxutil.LogRemoveListener(r.logDepth, base, r.id, name)
-
+func (r *Receiver) RemoveKey(name string, key ListenerKey) *Listener {
r.mtx.Lock()
defer r.mtx.Unlock()
- bl := r.bx.RemoveListener(base)
- delete(r.bls, bl)
-
- if bl != nil {
- r.wg.Done()
+ r.bx.RemoveKey(key)
+ bl := r.lm.RemoveKey(key)
+ if bl == nil {
+ return nil
}
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ r.wg.Done()
return bl
}
-func (r *Receiver) RemoveBaseListener(name string, base MsgBase) {
- r.removeListener(name, base)
-}
+func (r *Receiver) RemoveListener(name string, listener *Listener) *ListenerKey {
+ r.mtx.Lock()
+ defer r.mtx.Unlock()
-func (r *Receiver) RemoveSeqListener(name string, seq BleSeq) {
- base := MsgBase{
- Op: -1,
- Type: -1,
- Seq: seq,
- ConnHandle: -1,
+ r.bx.RemoveListener(listener)
+ key := r.lm.RemoveListener(listener)
+ if key == nil {
+ return nil
}
- r.removeListener(name, base)
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ r.wg.Done()
+ return key
}
func (r *Receiver) RemoveAll(name string) {
r.mtx.Lock()
- bls := r.bls
- r.bls = map[*Listener]MsgBase{}
+ bls := r.lm.ExtractAll()
r.mtx.Unlock()
- for _, base := range bls {
- r.removeListener(name, base)
+ for _, bl := range bls {
+ if key := r.bx.RemoveListener(bl); key != nil {
+ nmxutil.LogRemoveListener(r.logDepth, key, r.id, name)
+ }
+ r.wg.Done()
}
}
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmp/dispatch.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmp/dispatch.go
index 9051b5b..5f36b0b 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmp/dispatch.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmp/dispatch.go
@@ -127,6 +127,9 @@ func decodeRsp(pkt []byte) (NmpRsp, error) {
// Returns true if the response was dispatched.
func (d *Dispatcher) DispatchRsp(r NmpRsp) bool {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
+
log.Debugf("Received nmp rsp: %+v", r)
nl := d.seqListenerMap[r.Hdr().Seq]
@@ -142,9 +145,6 @@ func (d *Dispatcher) DispatchRsp(r NmpRsp) bool {
// Returns true if the response was dispatched.
func (d *Dispatcher) Dispatch(data []byte) bool {
- d.mtx.Lock()
- defer d.mtx.Unlock()
-
pkt := d.reassembler.RxFrag(data)
if pkt == nil {
return false
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmxutil/nmxutil.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmxutil/nmxutil.go
index 078a06d..84f45ae 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmxutil/nmxutil.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/nmxutil/nmxutil.go
@@ -10,9 +10,10 @@ import (
"sync"
"time"
+ log "github.com/Sirupsen/logrus"
"github.com/ugorji/go/codec"
- log "github.com/Sirupsen/logrus"
+ "mynewt.apache.org/newt/util"
)
const DURATION_FOREVER time.Duration = math.MaxInt64
@@ -45,6 +46,7 @@ func SetLogLevel(level log.Level) {
func Assert(cond bool) {
if Debug && !cond {
+ util.PrintStacks()
panic("Failed assertion")
}
}
@@ -115,16 +117,16 @@ func LogRemoveOicListener(parentLevel int, token []byte) {
fmt.Sprintf("token=%+v", token))
}
-func LogAddListener(parentLevel int, base interface{}, id uint32,
+func LogAddListener(parentLevel int, key interface{}, id uint32,
name string) {
LogListener(parentLevel, "add-ble-listener",
- fmt.Sprintf("[%d] %s: base=%+v", id, name, base))
+ fmt.Sprintf("[%d] %s: base=%+v", id, name, key))
}
-func LogRemoveListener(parentLevel int, base interface{}, id uint32,
+func LogRemoveListener(parentLevel int, key interface{}, id uint32,
name string) {
LogListener(parentLevel, "remove-ble-listener",
- fmt.Sprintf("[%d] %s: base=%+v", id, name, base))
+ fmt.Sprintf("[%d] %s: base=%+v", id, name, key))
}
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/oic/dispatch.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/oic/dispatch.go
index 810abe7..33f0ac9 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/oic/dispatch.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/oic/dispatch.go
@@ -125,25 +125,42 @@ func (d *Dispatcher) AddListener(token []byte) (*Listener, error) {
}
func (d *Dispatcher) RemoveListener(token []byte) *Listener {
- nmxutil.LogRemoveOicListener(d.logDepth, token)
-
d.mtx.Lock()
defer d.mtx.Unlock()
ot, err := NewToken(token)
if err != nil {
+ log.Errorf("Error creating OIC token: %s", err.Error())
return nil
}
ol := d.tokenListenerMap[ot]
- if ol != nil {
- ol.Close()
- delete(d.tokenListenerMap, ot)
+ if ol == nil {
+ return nil
}
+ nmxutil.LogRemoveOicListener(d.logDepth, token)
+ ol.Close()
+ delete(d.tokenListenerMap, ot)
+
return ol
}
+func (d *Dispatcher) dispatchRsp(ot Token, msg *coap.Message) bool {
+ d.mtx.Lock()
+ defer d.mtx.Unlock()
+
+ ol := d.tokenListenerMap[ot]
+
+ if ol == nil {
+ log.Printf("No listener for incoming OIC message; token=%#v", ot)
+ return false
+ }
+
+ ol.RspChan <- msg
+ return true
+}
+
// Returns true if the response was dispatched.
func (d *Dispatcher) Dispatch(data []byte) bool {
var msg *coap.Message
@@ -171,17 +188,7 @@ func (d *Dispatcher) Dispatch(data []byte) bool {
return false
}
- d.mtx.Lock()
- ol := d.tokenListenerMap[ot]
- d.mtx.Unlock()
-
- if ol == nil {
- log.Printf("No listener for incoming OIC message; token=%#v", ot)
- return false
- }
-
- ol.RspChan <- msg
- return true
+ return d.dispatchRsp(ot, msg)
}
func (d *Dispatcher) ErrorOne(token Token, err error) error {
diff --git a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/omp/dispatch.go b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/omp/dispatch.go
index 7647048..093efa5 100644
--- a/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/omp/dispatch.go
+++ b/newtmgr/vendor/mynewt.apache.org/newtmgr/nmxact/omp/dispatch.go
@@ -38,33 +38,33 @@ type Dispatcher struct {
}
func NewDispatcher(isTcp bool, logDepth int) (*Dispatcher, error) {
- r := &Dispatcher{
+ d := &Dispatcher{
nmpd: nmp.NewDispatcher(logDepth + 1),
oicd: oic.NewDispatcher(isTcp, logDepth+1),
stopCh: make(chan struct{}),
}
// Listen for OMP responses. This should never fail.
- if err := r.addOmpListener(); err != nil {
+ if err := d.addOmpListener(); err != nil {
log.Errorf("Unexpected failure to add OMP listener: " + err.Error())
return nil, err
}
- return r, nil
+ return d, nil
}
-func (r *Dispatcher) addOmpListener() error {
+func (d *Dispatcher) addOmpListener() error {
// OMP responses are identifiable by the lack of a CoAP token. Set up a
// permanent listener to receive these messages.
- ol, err := r.AddOicListener(nil)
+ ol, err := d.AddOicListener(nil)
if err != nil {
return err
}
- r.wg.Add(1)
+ d.wg.Add(1)
go func() {
- defer r.RemoveOicListener(nil)
- defer r.wg.Done()
+ defer d.RemoveOicListener(nil)
+ defer d.wg.Done()
for {
select {
@@ -73,13 +73,13 @@ func (r *Dispatcher) addOmpListener() error {
if err != nil {
log.Debugf("OMP decode failure: %s", err.Error())
} else {
- r.nmpd.DispatchRsp(rsp)
+ d.nmpd.DispatchRsp(rsp)
}
case err := <-ol.ErrChan:
log.Debugf("OIC error: %s", err.Error())
- case <-r.stopCh:
+ case <-d.stopCh:
return
}
}
@@ -88,36 +88,36 @@ func (r *Dispatcher) addOmpListener() error {
return nil
}
-func (r *Dispatcher) Stop() {
- r.stopCh <- struct{}{}
- r.wg.Wait()
+func (d *Dispatcher) Stop() {
+ d.stopCh <- struct{}{}
+ d.wg.Wait()
}
-func (r *Dispatcher) Dispatch(data []byte) bool {
- return r.oicd.Dispatch(data)
+func (d *Dispatcher) Dispatch(data []byte) bool {
+ return d.oicd.Dispatch(data)
}
-func (r *Dispatcher) AddOicListener(token []byte) (*oic.Listener, error) {
- return r.oicd.AddListener(token)
+func (d *Dispatcher) AddOicListener(token []byte) (*oic.Listener, error) {
+ return d.oicd.AddListener(token)
}
-func (r *Dispatcher) RemoveOicListener(token []byte) *oic.Listener {
- return r.oicd.RemoveListener(token)
+func (d *Dispatcher) RemoveOicListener(token []byte) *oic.Listener {
+ return d.oicd.RemoveListener(token)
}
-func (r *Dispatcher) AddNmpListener(seq uint8) (*nmp.Listener, error) {
- return r.nmpd.AddListener(seq)
+func (d *Dispatcher) AddNmpListener(seq uint8) (*nmp.Listener, error) {
+ return d.nmpd.AddListener(seq)
}
-func (r *Dispatcher) RemoveNmpListener(seq uint8) *nmp.Listener {
- return r.nmpd.RemoveListener(seq)
+func (d *Dispatcher) RemoveNmpListener(seq uint8) *nmp.Listener {
+ return d.nmpd.RemoveListener(seq)
}
-func (r *Dispatcher) ErrorOneNmp(seq uint8, err error) error {
- return r.nmpd.ErrorOne(seq, err)
+func (d *Dispatcher) ErrorOneNmp(seq uint8, err error) error {
+ return d.nmpd.ErrorOne(seq, err)
}
-func (r *Dispatcher) ErrorAll(err error) {
- r.nmpd.ErrorAll(err)
- r.oicd.ErrorAll(err)
+func (d *Dispatcher) ErrorAll(err error) {
+ d.nmpd.ErrorAll(err)
+ d.oicd.ErrorAll(err)
}
--
To stop receiving notification emails like this one, please contact
"commits@mynewt.apache.org" <co...@mynewt.apache.org>.