Skip to content

fix(interaction): handle MoreChunkedMessages in Client.Subscribe #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
p0fi merged 1 commit into from
Sep 16, 2026
Merged

fix(interaction): handle MoreChunkedMessages in Client.Subscribe #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 77 additions & 47 deletions internal/interaction/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,12 @@ func (c *Client) invokeInternal(ctx context.Context, session *protocol.Session,

// Subscribe sends a SubscribeRequest and returns a Subscription that delivers
// periodic attribute reports. The subscription runs in the background and
// delivers reports through the Subscription.Reports channel.
// delivers reports through the Subscription.Reports channel. Both the priming
// report and every ongoing report cycle handle chunked responses
// (MoreChunkedMessages) the same way Client.Read does: each ReportData
// message is acknowledged with StatusResponse(Success) as it arrives, and the
// AttributeReports are accumulated until MoreChunkedMessages is false or
// absent before the batch is treated as complete.
func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths []AttributePath, minInterval, maxInterval uint16) (*Subscription, error) {
exchange, err := c.exchangeManager.NewExchange(ctx, session)
if err != nil {
Expand All @@ -303,43 +308,56 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths
return nil, fmt.Errorf("interaction: sending subscribe request: %w", err)
}

// Wait for the initial ReportData (priming report).
msg, err := exchange.Receive(ctx)
if err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: receiving priming report: %w", err)
}
// Wait for the initial ReportData (priming report), accumulating across
// MoreChunkedMessages chunks the same way Client.Read does.
var primingReports []AttributeReport
for {
msg, err := exchange.Receive(ctx)
if err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: receiving priming report: %w", err)
}

if err := checkStatusResponse(msg); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, err
}
if err := checkStatusResponse(msg); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, err
}

if msg.Protocol.ProtocolOpcode != OpcodeReportData {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: unexpected opcode 0x%02X, want ReportData (0x%02X)",
msg.Protocol.ProtocolOpcode, OpcodeReportData)
}
if msg.Protocol.ProtocolOpcode != OpcodeReportData {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: unexpected opcode 0x%02X, want ReportData (0x%02X)",
msg.Protocol.ProtocolOpcode, OpcodeReportData)
}

var primingReport ReportData
if err := tlv.Unmarshal(msg.Payload, &primingReport); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: decoding priming report: %w", err)
}
primingData, primingErr := splitAttributeReports(primingReport.AttributeReports)
var report ReportData
if err := tlv.Unmarshal(msg.Payload, &report); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: decoding priming report: %w", err)
}

// Send StatusResponse(Success) to acknowledge the priming report. This
// happens regardless of whether the report carried a per-attribute error
// status — the ACK is a transport-level acknowledgment of receipt, not an
// endorsement of the report's content.
ack := StatusResponseMessage{
Status: uint8(StatusSuccess),
}
if err := sendIMMessage(ctx, exchange, OpcodeStatusResponse, ack); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: acknowledging priming report: %w", err)
primingReports = append(primingReports, report.AttributeReports...)

// Send StatusResponse(Success) to acknowledge this chunk. This
// happens regardless of whether the report carried a per-attribute
// error status or whether more chunks are coming — the ACK is a
// transport-level acknowledgment of receipt, not an endorsement of
// the report's content.
ack := StatusResponseMessage{
Status: uint8(StatusSuccess),
}
if err := sendIMMessage(ctx, exchange, OpcodeStatusResponse, ack); err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: acknowledging priming report: %w", err)
}

if report.MoreChunkedMessages == nil || !*report.MoreChunkedMessages {
break
}
slog.Debug("interaction: subscribe priming report chunked, requesting next chunk", "so_far", len(primingReports))
}

primingData, primingErr := splitAttributeReports(primingReports)

// A priming attribute-status failure must not be presented as a
// successful establishment.
if primingErr != nil {
Expand All @@ -348,7 +366,7 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths
}

// Wait for the SubscribeResponse.
msg, err = exchange.Receive(ctx)
msg, err := exchange.Receive(ctx)
if err != nil {
c.exchangeManager.CloseExchange(exchange)
return nil, fmt.Errorf("interaction: receiving subscribe response: %w", err)
Expand Down Expand Up @@ -385,12 +403,16 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths
reports <- primingData
}

// Run background goroutine to receive periodic reports.
// Run background goroutine to receive periodic reports. Each report
// cycle may itself span multiple ReportData messages (MoreChunkedMessages);
// accumulated is the buffer for the cycle currently being assembled.
go func() {
defer c.exchangeManager.CloseExchange(exchange)
defer close(reports)
defer close(errs)

var accumulated []AttributeReport

for {
msg, err := exchange.Receive(subCtx)
if err != nil {
Expand Down Expand Up @@ -431,20 +453,12 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths
return
}

dataReports, statusErr := splitAttributeReports(report.AttributeReports)

if len(dataReports) > 0 {
select {
case reports <- dataReports:
case <-subCtx.Done():
return
}
}
accumulated = append(accumulated, report.AttributeReports...)

// Acknowledge the report. This happens regardless of whether it
// carried a per-attribute error status — the ACK is a
// transport-level acknowledgment of receipt, not an endorsement
// of the report's content.
// Acknowledge this chunk. This happens regardless of whether it
// carried a per-attribute error status or whether more chunks
// are coming — the ACK is a transport-level acknowledgment of
// receipt, not an endorsement of the report's content.
reportAck := StatusResponseMessage{
Status: uint8(StatusSuccess),
}
Expand All @@ -456,6 +470,22 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths
return
}

if report.MoreChunkedMessages != nil && *report.MoreChunkedMessages {
slog.Debug("interaction: subscription report chunked, awaiting next chunk", "so_far", len(accumulated))
continue
}

dataReports, statusErr := splitAttributeReports(accumulated)
accumulated = nil

if len(dataReports) > 0 {
select {
case reports <- dataReports:
case <-subCtx.Done():
return
}
}

if statusErr != nil {
select {
case errs <- statusErr:
Expand Down
91 changes: 91 additions & 0 deletions internal/interaction/client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -522,6 +522,97 @@ func TestClient_Subscribe(t *testing.T) {
r.sub.Cancel()
}

// TestClient_Subscribe_ChunkedPrimingReport verifies that a priming report
// spanning multiple ReportData messages (MoreChunkedMessages) is accumulated
// across all chunks before being delivered as a single batch on Reports, and
// that each chunk is acknowledged with StatusResponse(Success).
func TestClient_Subscribe_ChunkedPrimingReport(t *testing.T) {
em := protocol.NewExchangeManager()
session := &protocol.Session{ID: 12, Type: protocol.SessionCASE}
client := NewClient(em)

ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()

type result struct {
sub *Subscription
err error
}
ch := make(chan result, 1)

go func() {
sub, err := client.Subscribe(ctx, session,
[]AttributePath{
NewAttributePath(1, 0x0006, 0x0000),
NewAttributePath(1, 0x0008, 0x0000),
},
1, 60,
)
ch <- result{sub, err}
}()

time.Sleep(50 * time.Millisecond)

// First chunk with MoreChunkedMessages=true.
more := true
chunk1 := ReportData{
MoreChunkedMessages: &more,
AttributeReports: []AttributeReport{
{
Data: &AttributeData{
DataVersion: 1,
Path: NewAttributePath(1, 0x0006, 0x0000),
Data: []byte{0x09},
},
},
},
}
injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1)

time.Sleep(50 * time.Millisecond)

// Second (final) chunk.
chunk2 := ReportData{
AttributeReports: []AttributeReport{
{
Data: &AttributeData{
DataVersion: 1,
Path: NewAttributePath(1, 0x0008, 0x0000),
Data: []byte{0x04, 0x80},
},
},
},
}
injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2)

time.Sleep(50 * time.Millisecond)

subResp := SubscribeResponse{
SubscriptionID: 43,
MaxInterval: 30,
}
injectResponse(t, em, session.ID, 0, OpcodeSubscribeResponse, subResp)

r := <-ch
if r.err != nil {
t.Fatalf("Subscribe: %v", r.err)
}
if r.sub == nil {
t.Fatal("subscription should not be nil")
}

select {
case reports := <-r.sub.Reports:
if len(reports) != 2 {
t.Fatalf("priming reports len = %d, want 2 (union of both chunks)", len(reports))
}
case <-time.After(1 * time.Second):
t.Fatal("timed out waiting for priming report")
}

r.sub.Cancel()
}

func TestClient_Subscribe_Error(t *testing.T) {
em := protocol.NewExchangeManager()
session := &protocol.Session{ID: 10, Type: protocol.SessionCASE}
Expand Down
Loading