From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 21:49:08 +0000 (-0300) Subject: Dispatch parsed addresses without topic retries X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/10c345f1d112061688409cef7710b4ac14a966aa?hp=c376881c67694cd9931d80fb4a04ec69751d222b Dispatch parsed addresses without topic retries Incoming publication dispatch passes its validated address to the shared payload parser, so each topic is parsed once. Typed parsers keep their operation, envelope, and payload checks. Their tests cover valid data, wrong operations, and invalid envelopes without changing the existing wire contract or rejected inputs. Issue: 8fa2f25 --- diff --git a/doc/mqtt.rst b/doc/mqtt.rst index d13015e..e0a2d62 100644 --- a/doc/mqtt.rst +++ b/doc/mqtt.rst @@ -34,7 +34,7 @@ contract as the parsers through shared typed validation. A builder encodes its validated payload without parsing it. Batch members use the same metadata and value validation without envelope encoding. Access flag presence checks apply to parsed JSON only. ``Parse`` selects the -typed payload and address kind from an incoming topic. A caller does not +typed payload and address kind after one topic parse. A caller does not split or join topics and does not build an envelope. A ``Publication`` contains topic, JSON bytes, QoS, and retention. It diff --git a/protocol/builders_test.go b/protocol/builders_test.go index b2b5a4a..20bd547 100644 --- a/protocol/builders_test.go +++ b/protocol/builders_test.go @@ -239,3 +239,139 @@ func TestTypedParsersRejectWrongOperations(t *testing.T) { require.Error(t, parse(Root+"/devices/"+deviceID+"/unknown", []byte(`{}`), false)) } } + +func TestTypedParsersKeepValidationAfterTopicDispatch(t *testing.T) { + deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + metadata := Metadata{ + MessageID: messageID, + ExpiresAt: &deadline, + CorrelationID: commandID, + } + testCases := []struct { + name string + kind Kind + target string + payload any + parse func(string, []byte, bool) (any, error) + }{ + { + name: "descriptor keeps its machine metadata", + kind: Descriptor, + payload: testDescriptor(), + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseDescriptor(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "availability keeps its reachability", + kind: Availability, + payload: AvailabilityPayload{Status: "online"}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseAvailability(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "reported keeps its complete value", + kind: Reported, + target: "relay", + payload: ValuePayload{Value: json.RawMessage(`true`)}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseReported(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "telemetry keeps its sample", + kind: Telemetry, + target: "power", + payload: ValuePayload{Value: json.RawMessage(`12`)}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseTelemetry(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "event keeps its open fields", + kind: Event, + target: "press", + payload: EventPayload{"action": json.RawMessage(`"press"`)}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseEvent(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "command keeps its requested value", + kind: Command, + target: "relay", + payload: ValuePayload{Value: json.RawMessage(`true`)}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseCommand(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "command batch keeps its member order", + kind: CommandBatch, + payload: testBatch(deadline), + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseCommandBatch(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "result keeps its correlated outcome", + kind: Result, + target: commandID, + payload: ResultPayload{Status: "completed"}, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseResult(topic, data, retained) + return parsed.Payload, err + }, + }, + { + name: "hub status keeps its snapshot", + kind: HubStatus, + payload: HubStatusPayload{ + ID: deviceID, + Address: "example.com", + Status: "ready", + }, + parse: func(topic string, data []byte, retained bool) (any, error) { + parsed, err := ParseHubStatus(topic, data, retained) + return parsed.Payload, err + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + topic, err := (Address{ + Kind: tc.kind, + ID: deviceID, + Target: tc.target, + }).topic() + require.NoError(t, err) + + data, err := encodeEnvelope(metadata, tc.payload) + require.NoError(t, err) + + parsed, err := tc.parse(topic, data, false) + require.NoError(t, err) + require.Equal(t, tc.payload, parsed) + + _, err = tc.parse(topic, []byte(`{}`), false) + require.Error(t, err) + + wrongTopic := Root + "/devices/" + deviceID + "/availability" + if tc.kind == Availability { + wrongTopic = Root + "/devices/" + deviceID + "/descriptor" + } + + _, err = tc.parse(wrongTopic, data, false) + require.Error(t, err) + }) + } +} diff --git a/protocol/message.go b/protocol/message.go index cb575d2..50a2e1c 100644 --- a/protocol/message.go +++ b/protocol/message.go @@ -42,41 +42,28 @@ func Parse(topic string, data []byte, retained bool) (Message[any], error) { switch address.Kind { case Descriptor: - return eraseMessage(ParseDescriptor(topic, data, retained)) + return eraseMessage(parsePayload[DescriptorPayload](address, data, retained)) case Availability: - return eraseMessage(ParseAvailability(topic, data, retained)) + return eraseMessage(parsePayload[AvailabilityPayload](address, data, retained)) case Reported: - return eraseMessage(ParseReported(topic, data, retained)) + return eraseMessage(parsePayload[ValuePayload](address, data, retained)) case Telemetry: - return eraseMessage(ParseTelemetry(topic, data, retained)) + return eraseMessage(parsePayload[ValuePayload](address, data, retained)) case Event: - return eraseMessage(ParseEvent(topic, data, retained)) + return eraseMessage(parsePayload[EventPayload](address, data, retained)) case Command: - return eraseMessage(ParseCommand(topic, data, retained)) + return eraseMessage(parsePayload[ValuePayload](address, data, retained)) case CommandBatch: - return eraseMessage(ParseCommandBatch(topic, data, retained)) + return eraseMessage(parsePayload[BatchPayload](address, data, retained)) case Result: - return eraseMessage(ParseResult(topic, data, retained)) + return eraseMessage(parsePayload[ResultPayload](address, data, retained)) case HubStatus: - return eraseMessage(ParseHubStatus(topic, data, retained)) + return eraseMessage(parsePayload[HubStatusPayload](address, data, retained)) default: return Message[any]{}, fmt.Errorf("unsupported message kind") } } -func eraseMessage[T any](message Message[T], err error) (Message[any], error) { - if err != nil { - return Message[any]{}, fmt.Errorf("parse message: %w", err) - } - - return Message[any]{ - Address: message.Address, - Metadata: message.Metadata, - Payload: message.Payload, - Retained: message.Retained, - }, nil -} - func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) { topic, err := address.topic() if err != nil { @@ -156,6 +143,19 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M return message, nil } +func eraseMessage[T any](message Message[T], err error) (Message[any], error) { + if err != nil { + return Message[any]{}, fmt.Errorf("parse message: %w", err) + } + + return Message[any]{ + Address: message.Address, + Metadata: message.Metadata, + Payload: message.Payload, + Retained: message.Retained, + }, nil +} + func parsePayload[T any](address Address, data []byte, retained bool) (Message[T], error) { kind := address.Kind @@ -252,28 +252,6 @@ func validatePayload(address Address, metadata Metadata, payload any) error { return nil } -func validateDescriptor(payload DescriptorPayload) error { - if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(string([]rune(payload.Attribution))) > 2000 || len(payload.Capabilities) == 0 { - return fmt.Errorf("descriptor metadata or capabilities are invalid") - } - - seen := make(map[string]bool, len(payload.Capabilities)) - - for _, capability := range payload.Capabilities { - if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] { - return fmt.Errorf("capability IDs and types must be valid and unique") - } - - if !capability.ReportsState && !capability.AcceptsCommands { - return fmt.Errorf("capability must have at least one access mode") - } - - seen[capability.ID] = true - } - - return nil -} - func validateDescriptorFlags(raw json.RawMessage) error { var fields struct { Capabilities []map[string]json.RawMessage `json:"capabilities"` @@ -296,6 +274,28 @@ func validateDescriptorFlags(raw json.RawMessage) error { return nil } +func validateDescriptor(payload DescriptorPayload) error { + if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(string([]rune(payload.Attribution))) > 2000 || len(payload.Capabilities) == 0 { + return fmt.Errorf("descriptor metadata or capabilities are invalid") + } + + seen := make(map[string]bool, len(payload.Capabilities)) + + for _, capability := range payload.Capabilities { + if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] { + return fmt.Errorf("capability IDs and types must be valid and unique") + } + + if !capability.ReportsState && !capability.AcceptsCommands { + return fmt.Errorf("capability must have at least one access mode") + } + + seen[capability.ID] = true + } + + return nil +} + func validateAvailability(payload AvailabilityPayload) error { if payload.Status != "online" && payload.Status != "offline" { return fmt.Errorf("availability must be online or offline")