// 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)
}
}
func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
- topic, err := address.Topic()
+ topic, err := address.topic()
if err != nil {
return Publication{}, fmt.Errorf("build address: %w", err)
}
- data, err := EncodeEnvelope(metadata, payload)
+ data, err := encodeEnvelope(metadata, payload)
if err != nil {
return Publication{}, fmt.Errorf("build envelope: %w", err)
}
Kind: Result,
ID: address.ID,
Target: metadata.MessageID,
- }).Topic()
+ }).topic()
if err != nil {
return Publication{}, fmt.Errorf("build command response topic: %w", err)
}
}
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)
}
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)
}
}, nil
}
-func validatePayload(address Address, envelope Envelope, payload any) error {
+func validatePayload(address Address, envelope wireEnvelope, payload any) error {
switch value := payload.(type) {
case DescriptorPayload:
err := validateDescriptor(value, envelope.Payload)
return nil
}
-func validateBatch(payload BatchPayload, envelope Envelope) error {
+func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
}
return fmt.Errorf("batch targets and IDs must be valid and distinct")
}
- _, err := EncodeEnvelope(command.Metadata, command.Payload)
+ _, err := encodeEnvelope(command.Metadata, command.Payload)
if err != nil {
return fmt.Errorf("validate batch command envelope: %w", err)
}
return nil
}
-func validateHub(payload HubStatusPayload, address Address, envelope Envelope) error {
+func validateHub(payload HubStatusPayload, address Address, envelope wireEnvelope) error {
if payload.ID != address.ID || envelope.ExpiresAt == nil {
return fmt.Errorf("hub status requires a matching ID and an expiry")
}