Exchange bounded M31 Modbus TCP requests

[onehouse] / protocol / message.go

diff --git a/protocol/message.go b/protocol/message.go
index 94ea7b582ee25fc252fe9dad504af018c4d51bc9..50a2e1ce22ea34c893af7d0d550a04412c38378e 100644 (file)
--- a/protocol/message.go
+++ b/protocol/message.go
@@ -35,64 +35,53 @@ type Publication struct {
 // EventPayload, BatchPayload, ResultPayload, or HubStatusPayload. Address.Kind
 // distinguishes command, reported, and telemetry value payloads.
 func Parse(topic string, data []byte, retained bool) (Message[any], error) {
 // EventPayload, BatchPayload, ResultPayload, or HubStatusPayload. Address.Kind
 // distinguishes command, reported, and telemetry value payloads.
 func Parse(topic string, data []byte, retained bool) (Message[any], error) {

-       address, err := ParseTopic(topic)
+       address, err := parseTopic(topic)

        if err != nil {
                return Message[any]{}, fmt.Errorf("parse address: %w", err)
        }
 
        switch address.Kind {
        case Descriptor:
        if err != nil {
                return Message[any]{}, fmt.Errorf("parse address: %w", err)
        }
 
        switch address.Kind {
        case Descriptor:

-               return eraseMessage(ParseDescriptor(topic, data, retained))
+               return eraseMessage(parsePayload[DescriptorPayload](address, data, retained))

        case Availability:
        case Availability:

-               return eraseMessage(ParseAvailability(topic, data, retained))
+               return eraseMessage(parsePayload[AvailabilityPayload](address, data, retained))

        case Reported:
        case Reported:

-               return eraseMessage(ParseReported(topic, data, retained))
+               return eraseMessage(parsePayload[ValuePayload](address, data, retained))

        case Telemetry:
        case Telemetry:

-               return eraseMessage(ParseTelemetry(topic, data, retained))
+               return eraseMessage(parsePayload[ValuePayload](address, data, retained))

        case Event:
        case Event:

-               return eraseMessage(ParseEvent(topic, data, retained))
+               return eraseMessage(parsePayload[EventPayload](address, data, retained))

        case Command:
        case Command:

-               return eraseMessage(ParseCommand(topic, data, retained))
+               return eraseMessage(parsePayload[ValuePayload](address, data, retained))

        case CommandBatch:
        case CommandBatch:

-               return eraseMessage(ParseCommandBatch(topic, data, retained))
+               return eraseMessage(parsePayload[BatchPayload](address, data, retained))

        case Result:
        case Result:

-               return eraseMessage(ParseResult(topic, data, retained))
+               return eraseMessage(parsePayload[ResultPayload](address, data, retained))

        case HubStatus:
        case HubStatus:

-               return eraseMessage(ParseHubStatus(topic, data, retained))
+               return eraseMessage(parsePayload[HubStatusPayload](address, data, retained))

        default:
                return Message[any]{}, fmt.Errorf("unsupported message kind")
        }
 }
 
        default:
                return Message[any]{}, fmt.Errorf("unsupported message kind")
        }
 }
 

-func eraseMessage[T any](message Message[T], err error) (Message[any], error) {
+func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
+       topic, err := address.topic()

        if err != nil {
        if err != nil {

-               return Message[any]{}, fmt.Errorf("parse message: %w", err)
+               return Publication{}, fmt.Errorf("build address: %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()
+       normalized := normalizePayload(payload)
+       err = validatePayload(address, metadata, normalized)

        if err != nil {
        if err != nil {

-               return Publication{}, fmt.Errorf("build address: %w", err)
+               return Publication{}, fmt.Errorf("validate publication: %w", err)

        }
 
        }
 

-       data, err := EncodeEnvelope(metadata, payload)
+       data, err := encodeEnvelope(metadata, normalized)

        if err != nil {
                return Publication{}, fmt.Errorf("build envelope: %w", err)
        }
 
        retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
        if err != nil {
                return Publication{}, fmt.Errorf("build envelope: %w", err)
        }
 
        retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus

-       _, err = parseMessage[T](address.Kind, topic, data, retain)
-       if err != nil {
-               return Publication{}, fmt.Errorf("validate publication: %w", err)
-       }

 
        qos := byte(1)
        if address.Kind == Telemetry {
 
        qos := byte(1)
        if address.Kind == Telemetry {


@@ -114,7 +103,7 @@ func buildMessage[T any](address Address, metadata Metadata, payload T) (Publica
                        Kind:   Result,
                        ID:     address.ID,
                        Target: metadata.MessageID,
                        Kind:   Result,
                        ID:     address.ID,
                        Target: metadata.MessageID,

-               }).Topic()
+               }).topic()

                if err != nil {
                        return Publication{}, fmt.Errorf("build command response topic: %w", err)
                }
                if err != nil {
                        return Publication{}, fmt.Errorf("build command response topic: %w", err)
                }


@@ -137,7 +126,7 @@ func buildMessage[T any](address Address, metadata Metadata, payload T) (Publica
 }
 
 func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) {
 }
 
 func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) {

-       address, err := ParseTopic(topic)
+       address, err := parseTopic(topic)

        if err != nil {
                return Message[T]{}, fmt.Errorf("parse address: %w", err)
        }
        if err != nil {
                return Message[T]{}, fmt.Errorf("parse address: %w", err)
        }


@@ -146,11 +135,35 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
                return Message[T]{}, fmt.Errorf("message topic does not match parser")
        }
 
                return Message[T]{}, fmt.Errorf("message topic does not match parser")
        }
 

+       message, err := parsePayload[T](address, data, retained)
+       if err != nil {
+               return Message[T]{}, fmt.Errorf("parse payload: %w", err)
+       }
+
+       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
+

        if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
                return Message[T]{}, fmt.Errorf("message must not be retained")
        }
 
        if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
                return Message[T]{}, fmt.Errorf("message must not be retained")
        }
 

-       envelope, err := ParseEnvelope(data)
+       envelope, err := parseEnvelope(data)

        if err != nil {
                return Message[T]{}, fmt.Errorf("parse envelope: %w", err)
        }
        if err != nil {
                return Message[T]{}, fmt.Errorf("parse envelope: %w", err)
        }


@@ -162,11 +175,18 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
                return Message[T]{}, fmt.Errorf("decode payload: %w", err)
        }
 
                return Message[T]{}, fmt.Errorf("decode payload: %w", err)
        }
 

-       err = validatePayload(address, envelope, payload)
+       err = validatePayload(address, envelope.Metadata, payload)

        if err != nil {
                return Message[T]{}, fmt.Errorf("validate payload: %w", err)
        }
 
        if err != nil {
                return Message[T]{}, fmt.Errorf("validate payload: %w", err)
        }
 

+       if address.Kind == Descriptor {
+               err = validateDescriptorFlags(envelope.Payload)
+               if err != nil {
+                       return Message[T]{}, fmt.Errorf("validate descriptor access flags: %w", err)
+               }
+       }
+

        return Message[T]{
                Address:  address,
                Metadata: envelope.Metadata,
        return Message[T]{
                Address:  address,
                Metadata: envelope.Metadata,


@@ -175,66 +195,64 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
        }, nil
 }
 
        }, nil
 }
 

-func validatePayload(address Address, envelope Envelope, payload any) error {
+// normalizePayload keeps nil raw values as JSON null without changing caller
+// data.
+func normalizePayload(payload any) any {

        switch value := payload.(type) {
        switch value := payload.(type) {

-       case DescriptorPayload:
-               err := validateDescriptor(value, envelope.Payload)
-               if err != nil {
-                       return fmt.Errorf("validate descriptor: %w", err)
-               }
-       case AvailabilityPayload:
-               if value.Status != "online" && value.Status != "offline" {
-                       return fmt.Errorf("availability must be online or offline")
-               }

        case ValuePayload:
        case ValuePayload:

-               if !json.Valid(value.Value) {
-                       return fmt.Errorf("value is required and must be valid JSON")
+               if value.Value == nil {
+                       value.Value = json.RawMessage(`null`)

                }
 
                }
 

-               if address.Kind == Command && envelope.ExpiresAt == nil {
-                       return fmt.Errorf("command must have an expiry")
-               }
-       case EventPayload:
-               // An event object has application-defined fields.
-       case ResultPayload:
-               if envelope.CorrelationID != address.Target {
-                       return fmt.Errorf("result correlation must match its command ID")
-               }
+               return value
+       case BatchPayload:
+               commands := make([]BatchCommand, len(value.Commands))
+               copy(commands, value.Commands)

 
 

-               switch value.Status {
-               case "accepted", "completed":
-                       if value.Reason != "" {
-                               return fmt.Errorf("successful result must not have a reason")
+               for index := range commands {
+                       if commands[index].Payload.Value == nil {
+                               commands[index].Payload.Value = json.RawMessage(`null`)

                        }
                        }

-               case "rejected", "failed":
-                       if value.Reason == "" {
-                               return fmt.Errorf("unsuccessful result must have a reason")
-                       }
-               default:
-                       return fmt.Errorf("result status is invalid")

                }
                }

+
+               value.Commands = commands
+
+               return value
+       default:
+               return payload
+       }
+}
+
+func validatePayload(address Address, metadata Metadata, payload any) error {
+       var err error
+
+       switch value := payload.(type) {
+       case DescriptorPayload:
+               err = validateDescriptor(value)
+       case AvailabilityPayload:
+               err = validateAvailability(value)
+       case ValuePayload:
+               err = validateValue(value, address.Kind, metadata)
+       case EventPayload:
+               err = validateEvent(value)
+       case ResultPayload:
+               err = validateResult(value, address, metadata)

        case BatchPayload:
        case BatchPayload:

-               err := validateBatch(value, envelope)
-               if err != nil {
-                       return fmt.Errorf("validate command batch: %w", err)
-               }
+               err = validateBatch(value, metadata)

        case HubStatusPayload:
        case HubStatusPayload:

-               err := validateHub(value, address, envelope)
-               if err != nil {
-                       return fmt.Errorf("validate hub status: %w", err)
-               }
+               err = validateHub(value, address, metadata)

        default:
                return fmt.Errorf("unsupported payload type")
        }
 
        default:
                return fmt.Errorf("unsupported payload type")
        }
 

+       if err != nil {
+               return fmt.Errorf("validate %s payload: %w", address.Kind, err)
+       }
+

        return nil
 }
 
        return nil
 }
 

-func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
-       if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(payload.Attribution) > 2000 || len(payload.Capabilities) == 0 {
-               return fmt.Errorf("descriptor metadata or capabilities are invalid")
-       }
-
+func validateDescriptorFlags(raw json.RawMessage) error {

        var fields struct {
                Capabilities []map[string]json.RawMessage `json:"capabilities"`
        }
        var fields struct {
                Capabilities []map[string]json.RawMessage `json:"capabilities"`
        }


@@ -244,9 +262,26 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
                return fmt.Errorf("read descriptor access flags: %w", err)
        }
 
                return fmt.Errorf("read descriptor access flags: %w", err)
        }
 

+       for _, capability := range fields.Capabilities {
+               for _, name := range []string{"reports_state", "accepts_commands"} {
+                       field := capability[name]
+                       if string(field) != "true" && string(field) != "false" {
+                               return fmt.Errorf("capability access flags must be present booleans")
+                       }
+               }
+       }
+
+       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))
 
        seen := make(map[string]bool, len(payload.Capabilities))
 

-       for index, capability := range 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 !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] {
                        return fmt.Errorf("capability IDs and types must be valid and unique")
                }


@@ -255,25 +290,73 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
                        return fmt.Errorf("capability must have at least one access mode")
                }
 
                        return fmt.Errorf("capability must have at least one access mode")
                }
 

-               for _, name := range []string{"reports_state", "accepts_commands"} {
-                       field := fields.Capabilities[index][name]
-                       if string(field) != "true" && string(field) != "false" {
-                               return fmt.Errorf("capability access flags must be present booleans")
-                       }
+               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")
+       }
+
+       return nil
+}
+
+func validateValue(payload ValuePayload, kind Kind, metadata Metadata) error {
+       if !json.Valid(payload.Value) {
+               return fmt.Errorf("value is required and must be valid JSON")
+       }
+
+       if kind == Command && metadata.ExpiresAt == nil {
+               return fmt.Errorf("command must have an expiry")
+       }
+
+       return nil
+}
+
+func validateEvent(payload EventPayload) error {
+       if payload == nil {
+               return fmt.Errorf("event must be a JSON object")
+       }
+
+       for _, value := range payload {
+               if value != nil && !json.Valid(value) {
+                       return fmt.Errorf("event values must be valid JSON")

                }
                }

+       }

 
 

-               seen[capability.ID] = true
+       return nil
+}
+
+func validateResult(payload ResultPayload, address Address, metadata Metadata) error {
+       if metadata.CorrelationID != address.Target {
+               return fmt.Errorf("result correlation must match its command ID")
+       }
+
+       switch payload.Status {
+       case "accepted", "completed":
+               if payload.Reason != "" {
+                       return fmt.Errorf("successful result must not have a reason")
+               }
+       case "rejected", "failed":
+               if payload.Reason == "" {
+                       return fmt.Errorf("unsuccessful result must have a reason")
+               }
+       default:
+               return fmt.Errorf("result status is invalid")

        }
 
        return nil
 }
 
        }
 
        return nil
 }
 

-func validateBatch(payload BatchPayload, envelope Envelope) error {
+func validateBatch(payload BatchPayload, metadata Metadata) error {

        if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
                return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
        }
 
        if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
                return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
        }
 

-       if envelope.MessageID != payload.Commands[0].MessageID {
+       if metadata.MessageID != payload.Commands[0].MessageID {

                return fmt.Errorf("batch ID must equal its first command ID")
        }
 
                return fmt.Errorf("batch ID must equal its first command ID")
        }
 


@@ -285,13 +368,14 @@ func validateBatch(payload BatchPayload, envelope Envelope) error {
                        return fmt.Errorf("batch targets and IDs must be valid and distinct")
                }
 
                        return fmt.Errorf("batch targets and IDs must be valid and distinct")
                }
 

-               _, err := EncodeEnvelope(command.Metadata, command.Payload)
+               err := validateMetadata(command.Metadata)

                if err != nil {
                if err != nil {

-                       return fmt.Errorf("validate batch command envelope: %w", err)
+                       return fmt.Errorf("validate batch command metadata: %w", err)

                }
 
                }
 

-               if command.ExpiresAt == nil || !json.Valid(command.Payload.Value) {
-                       return fmt.Errorf("batch commands require expiry and a JSON value")
+               err = validateValue(command.Payload, Command, command.Metadata)
+               if err != nil {
+                       return fmt.Errorf("validate batch command value: %w", err)

                }
 
                ids[command.MessageID] = true
                }
 
                ids[command.MessageID] = true


@@ -301,8 +385,8 @@ func validateBatch(payload BatchPayload, envelope Envelope) error {
        return nil
 }
 
        return nil
 }
 

-func validateHub(payload HubStatusPayload, address Address, envelope Envelope) error {
-       if payload.ID != address.ID || envelope.ExpiresAt == nil {
+func validateHub(payload HubStatusPayload, address Address, metadata Metadata) error {
+       if payload.ID != address.ID || metadata.ExpiresAt == nil {

                return fmt.Errorf("hub status requires a matching ID and an expiry")
        }
 
                return fmt.Errorf("hub status requires a matching ID and an expiry")
        }