diff --git a/internal/interaction/client.go b/internal/interaction/client.go index 5befab5..72ee3f3 100644 --- a/internal/interaction/client.go +++ b/internal/interaction/client.go @@ -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 { @@ -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 { @@ -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) @@ -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 { @@ -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), } @@ -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: diff --git a/internal/interaction/client_test.go b/internal/interaction/client_test.go index 4482b49..7b08bb0 100644 --- a/internal/interaction/client_test.go +++ b/internal/interaction/client_test.go @@ -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} diff --git a/internal/interaction/extra_test.go b/internal/interaction/extra_test.go index d5dd0ff..a74aa93 100644 --- a/internal/interaction/extra_test.go +++ b/internal/interaction/extra_test.go @@ -470,6 +470,188 @@ func TestClient_Subscribe_MultiplePeriodicReports(t *testing.T) { } } +// TestClient_Subscribe_ChunkedOngoingReport verifies that an ongoing report +// cycle spanning multiple ReportData messages (MoreChunkedMessages) is +// accumulated across all chunks and delivered as exactly one batch on +// Reports, with each chunk acknowledged as it arrives. +func TestClient_Subscribe_ChunkedOngoingReport(t *testing.T) { + em := protocol.NewExchangeManager() + session := &protocol.Session{ID: 36, Type: protocol.SessionCASE} + client := NewClient(em) + + sub := establishSubscription(t, em, session, client) + defer sub.Cancel() + + time.Sleep(60 * time.Millisecond) + + more := true + chunk1 := ReportData{ + MoreChunkedMessages: &more, + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 5, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x09}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1) + + time.Sleep(60 * time.Millisecond) + + chunk2 := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 5, + Path: NewAttributePath(1, 0x0008, 0x0000), + Data: []byte{0x04, 0x80}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 2 { + t.Fatalf("reports len = %d, want 2 (union of both chunks)", len(reports)) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for chunked ongoing report") + } + + // Confirm exactly one batch was delivered for the two-chunk cycle: a + // subsequent, unrelated report must arrive as its own separate batch, + // not be coalesced with (or duplicate) the chunked one above. + time.Sleep(60 * time.Millisecond) + nextReport := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 6, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x08}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, nextReport) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 1 { + t.Fatalf("next reports len = %d, want 1", len(reports)) + } + if reports[0].Data.DataVersion != 6 { + t.Errorf("DataVersion = %d, want 6", reports[0].Data.DataVersion) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for report after chunked cycle") + } +} + +// TestClient_Subscribe_ChunkedOngoingReport_EmptyAfterSplit verifies that a +// chunked report cycle whose accumulated AttributeReports contain no data and +// no error (e.g. every entry is a per-attribute Status of Success, which +// splitAttributeReports treats as neither data nor a terminal error) sends +// nothing on Reports or Errors for that cycle, and does not leak into or +// otherwise corrupt the next cycle's batch. +func TestClient_Subscribe_ChunkedOngoingReport_EmptyAfterSplit(t *testing.T) { + em := protocol.NewExchangeManager() + session := &protocol.Session{ID: 37, Type: protocol.SessionCASE} + client := NewClient(em) + + sub := establishSubscription(t, em, session, client) + defer sub.Cancel() + + time.Sleep(60 * time.Millisecond) + + more := true + chunk1 := ReportData{ + MoreChunkedMessages: &more, + AttributeReports: []AttributeReport{ + { + Status: &AttributeStatus{ + Path: NewAttributePath(1, 0x0006, 0x0000), + Status: StatusIB{Status: uint8(StatusSuccess)}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1) + + time.Sleep(60 * time.Millisecond) + + chunk2 := ReportData{ + AttributeReports: []AttributeReport{ + { + Status: &AttributeStatus{ + Path: NewAttributePath(1, 0x0008, 0x0000), + Status: StatusIB{Status: uint8(StatusSuccess)}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2) + + // Neither Reports nor Errors should fire for this cycle. + select { + case reports, ok := <-sub.Reports: + if ok { + t.Fatalf("expected no report for an empty-after-split cycle, got: %+v", reports) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(300 * time.Millisecond): + // No send, as expected. + } + + // The next cycle must still be delivered as its own clean batch, proving + // the empty cycle didn't leave stale data in the accumulation buffer. + nextReport := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 7, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x09}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, nextReport) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 1 { + t.Fatalf("next reports len = %d, want 1", len(reports)) + } + if reports[0].Data.DataVersion != 7 { + t.Errorf("DataVersion = %d, want 7", reports[0].Data.DataVersion) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for report after empty-after-split cycle") + } +} + // TestClient_Subscribe_CancelDrainsChannels verifies that calling Cancel // causes both the Reports and Errors channels to be closed. func TestClient_Subscribe_CancelDrainsChannels(t *testing.T) {