events use lowercase ASCII letters and digits, with single internal
hyphen or period separators. Topic builders reject invalid addresses.
-``Address.Topic`` and ``ParseTopic`` cover device descriptor,
-availability, reported values, telemetry, events, commands, command
-batches, results, and Zigbee hub status. Subscription helpers supply the
-core input filters, each device's command filter, and hub status filter.
-The command filter includes the reserved ``$batch`` target.
-
-``EncodeEnvelope`` and ``ParseEnvelope`` validate schema, IDs, optional
-UTC timestamps, and a required object payload. Unknown envelope fields,
-trailing JSON, unsupported schema versions, and malformed topics are
-errors. A receiver checks expiry against its own clock. The message ID
-stays unchanged on retry. Its timestamp is not the observation time.
+Private topic helpers cover device descriptor, availability, reported
+values, telemetry, events, commands, command batches, results, and
+Zigbee hub status. Subscription helpers supply the core input filters,
+each device's command filter, and hub status filter. The command filter
+includes the reserved ``$batch`` target.
+
+Private envelope helpers validate schema, IDs, optional UTC timestamps,
+and a required object payload. Unknown envelope fields, trailing JSON,
+unsupported schema versions, and malformed topics are errors. A receiver
+checks expiry against its own clock. The message ID stays unchanged on
+retry. Its timestamp is not the observation time.
Measurements and online availability repeat at five-minute intervals.
The device contact timeout is fifteen minutes. These constants do not
Target string
}
-// Topic validates the address and returns its exact publication topic.
-func (a Address) Topic() (string, error) {
+// topic validates the address and returns its exact publication topic.
+func (a Address) topic() (string, error) {
if !ValidID(a.ID) {
return "", fmt.Errorf("address ID must be a canonical ULID")
}
return Root + "/devices/" + a.ID + "/" + string(a.Kind) + "/" + a.Target, nil
}
-// ParseTopic rejects malformed topics, wildcards, and unsupported operations.
-func ParseTopic(topic string) (Address, error) {
+// parseTopic rejects malformed topics, wildcards, and unsupported operations.
+func parseTopic(topic string) (Address, error) {
parts := strings.Split(topic, "/")
if len(parts) < 5 || parts[0]+"/"+parts[1] != Root {
return Address{}, fmt.Errorf("topic must use the versioned root")
return Address{}, fmt.Errorf("topic layout is invalid")
}
- canonical, err := address.Topic()
+ canonical, err := address.topic()
if err != nil {
return Address{}, fmt.Errorf("validate topic address: %w", err)
}
ID: deviceID,
Target: tc.target,
}
- topic, err := address.Topic()
+ topic, err := address.topic()
require.NoError(t, err)
expected := Root + "/devices/" + deviceID + "/" + tc.suffix
require.Equal(t, expected, topic)
- parsed, err := ParseTopic(topic)
+ parsed, err := parseTopic(topic)
require.NoError(t, err)
require.Equal(t, address, parsed)
})
}
for _, tc := range testCases {
- _, err := ParseTopic(tc)
+ _, err := parseTopic(tc)
require.Error(t, err, tc)
}
}
for _, command := range b.Commands {
commands = append(commands, batchWireCommand{
- Schema: Schema,
-
- Metadata: command.Metadata,
-
+ Schema: Schema,
+ Metadata: command.Metadata,
Capability: command.Capability,
-
- Payload: command.Payload,
+ Payload: command.Payload,
})
}
}
commands = append(commands, BatchCommand{
- Metadata: command.Metadata,
-
+ Metadata: command.Metadata,
Capability: command.Capability,
-
- Payload: command.Payload,
+ Payload: command.Payload,
})
}
return message, nil
}
-// BuildCommandBatch returns a validated commandbatch publication with its
+// BuildCommandBatch returns a validated command batch publication with its
// delivery rules.
func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload) (Publication, error) {
return buildMessage(Address{
func ParseCommandBatch(topic string, data []byte, retained bool) (Message[BatchPayload], error) {
message, err := parseMessage[BatchPayload](CommandBatch, topic, data, retained)
if err != nil {
- return Message[BatchPayload]{}, fmt.Errorf("parse commandbatch: %w", err)
+ return Message[BatchPayload]{}, fmt.Errorf("parse command batch: %w", err)
}
return message, nil
return message, nil
}
-// BuildHubStatus returns a validated hubstatus publication with its delivery
+// BuildHubStatus returns a validated hub status publication with its delivery
// rules.
func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) (Publication, error) {
return buildMessage(Address{
func ParseHubStatus(topic string, data []byte, retained bool) (Message[HubStatusPayload], error) {
message, err := parseMessage[HubStatusPayload](HubStatus, topic, data, retained)
if err != nil {
- return Message[HubStatusPayload]{}, fmt.Errorf("parse hubstatus: %w", err)
+ return Message[HubStatusPayload]{}, fmt.Errorf("parse hub status: %w", err)
}
return message, nil
}
if tc.kind == Command {
- response, err := ParseTopic(publication.ResponseTopic)
+ response, err := parseTopic(publication.ResponseTopic)
require.NoError(t, err)
require.Equal(t, Result, response.Kind)
require.Equal(t, messageID, response.Target)
// 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")
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- data, err := EncodeEnvelope(metadata, json.RawMessage(tc.payload))
+ data, err := encodeEnvelope(metadata, json.RawMessage(tc.payload))
if tc.kind == Event {
require.Error(t, err)
Kind: tc.kind,
ID: deviceID,
Target: tc.target,
- }).Topic()
+ }).topic()
require.NoError(t, err)
_, err = Parse(topic, data, false)
--- /dev/null
+package protocol
+
+import "time"
+
+const (
+ // MeasurementRepeatInterval bounds equal retained measurement repeats.
+ MeasurementRepeatInterval = 5 * time.Minute
+
+ // OnlineRepeatInterval bounds equal online availability repeats.
+ OnlineRepeatInterval = 5 * time.Minute
+
+ // DeviceContactTimeout bounds the age of device contact evidence.
+ DeviceContactTimeout = 15 * time.Minute
+)
"time"
)
-const (
- // Schema is the supported envelope and payload schema version.
- Schema = 1
-
- // MeasurementRepeatInterval bounds equal retained measurement repeats.
- MeasurementRepeatInterval = 5 * time.Minute
-
- // OnlineRepeatInterval bounds equal online availability repeats.
- OnlineRepeatInterval = 5 * time.Minute
-
- // DeviceContactTimeout bounds the age of device contact evidence.
- DeviceContactTimeout = 15 * time.Minute
-)
+// Schema is the supported envelope and payload schema version.
+const Schema = 1
// Metadata holds message identity and optional UTC times. Retries keep the same
// ID. A receiver checks ExpiresAt against its clock before processing.
CausationID string `json:"causation_id,omitempty"`
}
-// Envelope carries one JSON object and its schema and metadata.
-type Envelope struct {
+// wireEnvelope carries one JSON object and its schema and metadata.
+type wireEnvelope struct {
Schema int `json:"schema"`
Metadata
Payload json.RawMessage `json:"payload"`
}
-// EncodeEnvelope validates metadata and encodes a schema-1 object payload.
-func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) {
+// encodeEnvelope validates metadata and encodes a schema-1 object payload.
+func encodeEnvelope(metadata Metadata, payload any) ([]byte, error) {
encoded, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("encode payload: %w", err)
}
- envelope := Envelope{
+ envelope := wireEnvelope{
Schema: Schema,
Metadata: metadata,
Payload: encoded,
return encoded, nil
}
-// ParseEnvelope rejects unknown fields, trailing JSON, unsupported schemas,
+// parseEnvelope rejects unknown fields, trailing JSON, unsupported schemas,
// invalid IDs, non-UTC times, and missing or non-object payloads.
-func ParseEnvelope(data []byte) (Envelope, error) {
- var envelope Envelope
+func parseEnvelope(data []byte) (wireEnvelope, error) {
+ var parsed wireEnvelope
- err := decodeObject(data, &envelope)
+ err := decodeObject(data, &parsed)
if err != nil {
- return Envelope{}, fmt.Errorf("decode envelope: %w", err)
+ return wireEnvelope{}, fmt.Errorf("decode envelope: %w", err)
}
- err = validateEnvelope(envelope)
+ err = validateEnvelope(parsed)
if err != nil {
- return Envelope{}, fmt.Errorf("validate envelope: %w", err)
+ return wireEnvelope{}, fmt.Errorf("validate envelope: %w", err)
}
- return envelope, nil
+ return parsed, nil
}
-func validateEnvelope(envelope Envelope) error {
+func validateEnvelope(envelope wireEnvelope) error {
if envelope.Schema != Schema {
return fmt.Errorf("unsupported schema %d", envelope.Schema)
}
CorrelationID: commandID,
CausationID: deviceID,
}
- data, err := EncodeEnvelope(metadata, map[string]bool{"value": true})
+ data, err := encodeEnvelope(metadata, map[string]bool{"value": true})
require.NoError(t, err)
- envelope, err := ParseEnvelope(data)
+ envelope, err := parseEnvelope(data)
require.NoError(t, err)
require.Equal(t, Schema, envelope.Schema)
require.Equal(t, metadata, envelope.Metadata)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- _, err := ParseEnvelope([]byte(tc.data))
+ _, err := parseEnvelope([]byte(tc.data))
require.Error(t, err)
})
}
- _, err := EncodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{})
+ _, err := encodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{})
require.Error(t, err)
- _, err = EncodeEnvelope(Metadata{MessageID: messageID}, make(chan int))
+ _, err = encodeEnvelope(Metadata{MessageID: messageID}, make(chan int))
require.Error(t, err)
- _, err = EncodeEnvelope(Metadata{MessageID: messageID}, []int{1})
+ _, err = encodeEnvelope(Metadata{MessageID: messageID}, []int{1})
require.Error(t, err)
}
projects / onehouse / commitdiff