This is an automated email from the ASF dual-hosted git repository. sjanc pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git
commit d4a9509200d4709968c00eb1b261ddfc77510293 Author: Szymon Janc <[email protected]> AuthorDate: Thu Oct 22 00:24:39 2026 +0200 nmxact/nmcoap: Fix loss of back-to-back CoAP-TCP messages Reassembler.RxFrag stored the unread remainder returned by coap.PullTcp but then unconditionally cleared it after extracting a message. If a received chunk contained the end of one message and the start (or all) of the next one, the second message was silently dropped, and with it the response a caller was waiting for, which then timed out. Keep the remainder after a successful pull. Since RxFrag returns only one message per call, also add Receiver.RxBuffered and make Dispatcher.Dispatch keep dispatching until no complete message is left in the buffer, so a second complete message is delivered right away instead of waiting for the next fragment. Only CoAP-TCP (BLE) buffers data; UDP behaviour is unchanged. Assisted-by: Claude:claude-opus-5-5 --- nmxact/nmcoap/dispatch.go | 16 ++++++++--- nmxact/nmcoap/dispatch_test.go | 65 ++++++++++++++++++++++++++++++++++++++++++ nmxact/nmcoap/frag.go | 5 +++- nmxact/nmcoap/frag_test.go | 50 ++++++++++++++++++++++++++++++++ nmxact/nmcoap/receiver.go | 15 ++++++++++ 5 files changed, 146 insertions(+), 5 deletions(-) diff --git a/nmxact/nmcoap/dispatch.go b/nmxact/nmcoap/dispatch.go index 4048cd9..c8a3e71 100644 --- a/nmxact/nmcoap/dispatch.go +++ b/nmxact/nmcoap/dispatch.go @@ -104,12 +104,20 @@ func (d *Dispatcher) RemoveListener(mc MsgCriteria) *Listener { // Returns true if the response was dispatched. func (d *Dispatcher) Dispatch(data []byte) bool { - // See if this fragment completes a packet. - msg := d.rxer.Rx(data) - if msg == nil { - return false + dispatched := false + + // See if this fragment completes a packet. With TCP, a fragment may + // complete more than one packet; dispatch all of them. + for msg := d.rxer.Rx(data); msg != nil; msg = d.rxer.RxBuffered() { + if d.dispatchMsg(msg) { + dispatched = true + } } + return dispatched +} + +func (d *Dispatcher) dispatchMsg(msg coap.Message) bool { d.mtx.Lock() defer d.mtx.Unlock() diff --git a/nmxact/nmcoap/dispatch_test.go b/nmxact/nmcoap/dispatch_test.go new file mode 100644 index 0000000..a8df620 --- /dev/null +++ b/nmxact/nmcoap/dispatch_test.go @@ -0,0 +1,65 @@ +/** + * 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 nmcoap + +import ( + "testing" +) + +// A single TCP chunk containing two complete messages must dispatch both +// of them immediately, not just the first one. +func TestDispatcherTcpBackToBack(t *testing.T) { + d := NewDispatcher(true, 0) + + l1, err := d.AddListener(MsgCriteria{Token: []byte{1}}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + l2, err := d.AddListener(MsgCriteria{Token: []byte{2}}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer d.RemoveListener(l1.Criteria) + defer d.RemoveListener(l2.Criteria) + + b1 := tcpMsg(t, []byte{1}, "", []byte("one")) + b2 := tcpMsg(t, []byte{2}, "", []byte("two")) + if !d.Dispatch(append(append([]byte{}, b1...), b2...)) { + t.Fatalf("Dispatch() = false, want true") + } + + for i, l := range []*Listener{l1, l2} { + select { + case m := <-l.RspChan: + if m.Token()[0] != byte(i+1) { + t.Fatalf("listener %d got token % x", i+1, m.Token()) + } + default: + t.Fatalf("listener %d did not receive its message", i+1) + } + } +} + +func TestDispatcherNoListener(t *testing.T) { + d := NewDispatcher(true, 0) + if d.Dispatch(tcpMsg(t, []byte{1}, "", nil)) { + t.Fatalf("Dispatch() = true without a listener") + } +} diff --git a/nmxact/nmcoap/frag.go b/nmxact/nmcoap/frag.go index a7371d7..673da9d 100644 --- a/nmxact/nmcoap/frag.go +++ b/nmxact/nmcoap/frag.go @@ -48,6 +48,9 @@ func (r *Reassembler) RxFrag(frag []byte) *coap.TcpMessage { return nil } - r.cur = nil + // Keep any trailing bytes; they are the start of the next message. + if len(r.cur) == 0 { + r.cur = nil + } return tm } diff --git a/nmxact/nmcoap/frag_test.go b/nmxact/nmcoap/frag_test.go index 7a3a3ff..7aefb98 100644 --- a/nmxact/nmcoap/frag_test.go +++ b/nmxact/nmcoap/frag_test.go @@ -159,3 +159,53 @@ func TestReceiverTcp(t *testing.T) { t.Fatalf("unexpected message %v", got) } } + +// Two messages received in a single buffer must both be returned, one per +// call. Previously the second message was silently dropped. +func TestReassemblerBackToBack(t *testing.T) { + b1 := tcpMsg(t, []byte{1}, "", []byte("first")) + b2 := tcpMsg(t, []byte{2}, "", []byte("second")) + b3 := tcpMsg(t, []byte{3}, "", []byte("third")) + + r := NewReassembler() + buf := append(append([]byte{}, b1...), b2...) + + checkTcpMsg(t, r.RxFrag(buf), []byte{1}, []byte("first")) + checkTcpMsg(t, r.RxFrag(nil), []byte{2}, []byte("second")) + if m := r.RxFrag(nil); m != nil { + t.Fatalf("unexpected third message: %v", m) + } + + // First message complete plus the start of the next one. + buf = append(append([]byte{}, b1...), b3[:3]...) + checkTcpMsg(t, r.RxFrag(buf), []byte{1}, []byte("first")) + checkTcpMsg(t, r.RxFrag(b3[3:]), []byte{3}, []byte("third")) + if m := r.RxFrag(nil); m != nil { + t.Fatalf("unexpected extra message: %v", m) + } +} + +func TestReceiverRxBuffered(t *testing.T) { + udp := NewReceiver(false) + if m := udp.RxBuffered(); m != nil { + t.Fatalf("UDP receiver returned buffered message %v", m) + } + + tcp := NewReceiver(true) + if m := tcp.RxBuffered(); m != nil { + t.Fatalf("empty TCP receiver returned message %v", m) + } + + b1 := tcpMsg(t, []byte{1}, "", nil) + b2 := tcpMsg(t, []byte{2}, "", nil) + if m := tcp.Rx(append(append([]byte{}, b1...), b2...)); m == nil { + t.Fatalf("expected first message") + } + m := tcp.RxBuffered() + if m == nil || m.Token()[0] != 2 { + t.Fatalf("expected second message, got %v", m) + } + if m := tcp.RxBuffered(); m != nil { + t.Fatalf("unexpected message %v", m) + } +} diff --git a/nmxact/nmcoap/receiver.go b/nmxact/nmcoap/receiver.go index c8587ee..0835535 100644 --- a/nmxact/nmcoap/receiver.go +++ b/nmxact/nmcoap/receiver.go @@ -57,3 +57,18 @@ func (r *Receiver) Rx(data []byte) coap.Message { return m } } + +// RxBuffered returns the next complete message that is already buffered +// from previously received data, or nil if there is none. Only TCP can +// buffer messages. +func (r *Receiver) RxBuffered() coap.Message { + if r.reassembler == nil { + return nil + } + + tm := r.reassembler.RxFrag(nil) + if tm == nil { + return nil + } + return tm +}
