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 d13015e2266fcb0758be5828ca9631c2fe89894d..e0a2d6246a23f19295d68db48162f0ababcae3da 100644 (file)
--- 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 b2b5a4a9a567235d6538036203df616aad13f07a..20bd547b1b4ece5b5170a2331c35867c090109ab 100644 (file)
--- 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 cb575d2eca521743e840b6ff823e169d7b3d9c0f..50a2e1ce22ea34c893af7d0d550a04412c38378e 100644 (file)
--- 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")