Keep topic and envelope construction private

Typed publications and subscription helpers own the wire layout. The
public API exposes parsed addresses but no low-level topic or envelope
construction, which the driver framework does not need. Timing constants
have their own file, and batch fields and operation names use the normal
layout and words without changing the wire contract.

Issue: 8fa2f25

diff --git a/doc/mqtt.rst b/doc/mqtt.rst
index 55a9a67ebc2eaaa4992224ce7eb447a16a00029a..1f5fa4a6ce1d912916922588a84c7b466e95756a 100644 (file)
--- a/doc/mqtt.rst
+++ b/doc/mqtt.rst
@@ -7,17 +7,17 @@ command, and message IDs use canonical uppercase ULIDs. Capabilities and
 events use lowercase ASCII letters and digits, with single internal
 hyphen or period separators. Topic builders reject invalid addresses.
 
 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
 
 Measurements and online availability repeat at five-minute intervals.
 The device contact timeout is fifteen minutes. These constants do not

diff --git a/protocol/address.go b/protocol/address.go
index fbd0d2686c0bc2d5684f95c1024073cf4b09a0a1..7786dfdfdb3d6c7f3f6bb38ddc255ff70b70d18f 100644 (file)
--- a/protocol/address.go
+++ b/protocol/address.go
@@ -48,8 +48,8 @@ type Address struct {
        Target string
 }
 
        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")
        }
        if !ValidID(a.ID) {
                return "", fmt.Errorf("address ID must be a canonical ULID")
        }


@@ -88,8 +88,8 @@ func (a Address) Topic() (string, error) {
        return Root + "/devices/" + a.ID + "/" + string(a.Kind) + "/" + a.Target, nil
 }
 
        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")
        parts := strings.Split(topic, "/")
        if len(parts) < 5 || parts[0]+"/"+parts[1] != Root {
                return Address{}, fmt.Errorf("topic must use the versioned root")


@@ -119,7 +119,7 @@ func ParseTopic(topic string) (Address, error) {
                return Address{}, fmt.Errorf("topic layout is invalid")
        }
 
                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)
        }
        if err != nil {
                return Address{}, fmt.Errorf("validate topic address: %w", err)
        }

diff --git a/protocol/address_test.go b/protocol/address_test.go
index eca76d7520b81a3d0875b02caf9e61fbcac8343a..6325b62e3b1fb16f2bdbd2433aa0d03547c95509 100644 (file)
--- a/protocol/address_test.go
+++ b/protocol/address_test.go
@@ -79,7 +79,7 @@ func TestAddressRoundTrip(t *testing.T) {
                                ID:     deviceID,
                                Target: tc.target,
                        }
                                ID:     deviceID,
                                Target: tc.target,
                        }

-                       topic, err := address.Topic()
+                       topic, err := address.topic()

                        require.NoError(t, err)
 
                        expected := Root + "/devices/" + deviceID + "/" + tc.suffix
                        require.NoError(t, err)
 
                        expected := Root + "/devices/" + deviceID + "/" + tc.suffix


@@ -89,7 +89,7 @@ func TestAddressRoundTrip(t *testing.T) {
 
                        require.Equal(t, expected, topic)
 
 
                        require.Equal(t, expected, topic)
 

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

                        require.NoError(t, err)
                        require.Equal(t, address, parsed)
                })
                        require.NoError(t, err)
                        require.Equal(t, address, parsed)
                })


@@ -113,7 +113,7 @@ func TestParseTopicRejectsMalformedAddresses(t *testing.T) {
        }
 
        for _, tc := range testCases {
        }
 
        for _, tc := range testCases {

-               _, err := ParseTopic(tc)
+               _, err := parseTopic(tc)

                require.Error(t, err, tc)
        }
 }
                require.Error(t, err, tc)
        }
 }

diff --git a/protocol/batch_payload.go b/protocol/batch_payload.go
index 36709398ca6ffdfd6c41ae1b21eed9f37a10ddd5..6e0f07be3ba73c8cf3541a983db0a87dce49d3fb 100644 (file)
--- a/protocol/batch_payload.go
+++ b/protocol/batch_payload.go
@@ -21,13 +21,10 @@ func (b BatchPayload) MarshalJSON() ([]byte, error) {
 
        for _, command := range b.Commands {
                commands = append(commands, batchWireCommand{
 
        for _, command := range b.Commands {
                commands = append(commands, batchWireCommand{

-                       Schema: Schema,
-
-                       Metadata: command.Metadata,
-
+                       Schema:     Schema,
+                       Metadata:   command.Metadata,

                        Capability: command.Capability,
                        Capability: command.Capability,

-
-                       Payload: command.Payload,
+                       Payload:    command.Payload,

                })
        }
 
                })
        }
 


@@ -56,11 +53,9 @@ func (b *BatchPayload) UnmarshalJSON(data []byte) error {
                }
 
                commands = append(commands, BatchCommand{
                }
 
                commands = append(commands, BatchCommand{

-                       Metadata: command.Metadata,
-
+                       Metadata:   command.Metadata,

                        Capability: command.Capability,
                        Capability: command.Capability,

-
-                       Payload: command.Payload,
+                       Payload:    command.Payload,

                })
        }
 
                })
        }
 

diff --git a/protocol/builders.go b/protocol/builders.go
index b9a39f77b712eb17cdb3a4a4b7c35093c6194738..16f94f83987d2133fa85311aa5479af83650599c 100644 (file)
--- a/protocol/builders.go
+++ b/protocol/builders.go
@@ -121,7 +121,7 @@ func ParseCommand(topic string, data []byte, retained bool) (Message[ValuePayloa
        return message, nil
 }
 
        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{
 // delivery rules.
 func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload) (Publication, error) {
        return buildMessage(Address{


@@ -135,7 +135,7 @@ func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload)
 func ParseCommandBatch(topic string, data []byte, retained bool) (Message[BatchPayload], error) {
        message, err := parseMessage[BatchPayload](CommandBatch, topic, data, retained)
        if err != nil {
 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


@@ -160,7 +160,7 @@ func ParseResult(topic string, data []byte, retained bool) (Message[ResultPayloa
        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{
 // rules.
 func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) (Publication, error) {
        return buildMessage(Address{


@@ -174,7 +174,7 @@ func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) (
 func ParseHubStatus(topic string, data []byte, retained bool) (Message[HubStatusPayload], error) {
        message, err := parseMessage[HubStatusPayload](HubStatus, topic, data, retained)
        if err != nil {
 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
        }
 
        return message, nil

diff --git a/protocol/builders_test.go b/protocol/builders_test.go
index 159cca2bc7314b6be70ccdb2e3c85e2750aaf23b..b2b5a4a9a567235d6538036203df616aad13f07a 100644 (file)
--- a/protocol/builders_test.go
+++ b/protocol/builders_test.go
@@ -182,7 +182,7 @@ func TestTypedPublicationsRoundTrip(t *testing.T) {
                        }
 
                        if tc.kind == Command {
                        }
 
                        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)
                                require.NoError(t, err)
                                require.Equal(t, Result, response.Kind)
                                require.Equal(t, messageID, response.Target)

diff --git a/protocol/message.go b/protocol/message.go
index 94ea7b582ee25fc252fe9dad504af018c4d51bc9..0fd4ec9195796a53d529112ed41d1d457b469471 100644 (file)
--- a/protocol/message.go
+++ b/protocol/message.go
@@ -35,7 +35,7 @@ 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)
        }
        if err != nil {
                return Message[any]{}, fmt.Errorf("parse address: %w", err)
        }


@@ -78,12 +78,12 @@ func eraseMessage[T any](message Message[T], err error) (Message[any], error) {
 }
 
 func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
 }
 
 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)
        }
 
        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)
        }
        if err != nil {
                return Publication{}, fmt.Errorf("build envelope: %w", err)
        }


@@ -114,7 +114,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 +137,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)
        }


@@ -150,7 +150,7 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
                return Message[T]{}, fmt.Errorf("message must not be retained")
        }
 
                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)
        }


@@ -175,7 +175,7 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
        }, nil
 }
 
        }, 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)
        switch value := payload.(type) {
        case DescriptorPayload:
                err := validateDescriptor(value, envelope.Payload)


@@ -268,7 +268,7 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
        return nil
 }
 
        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)
        }
        if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
                return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
        }


@@ -285,7 +285,7 @@ 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 := encodeEnvelope(command.Metadata, command.Payload)

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


@@ -301,7 +301,7 @@ func validateBatch(payload BatchPayload, envelope Envelope) error {
        return nil
 }
 
        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")
        }
        if payload.ID != address.ID || envelope.ExpiresAt == nil {
                return fmt.Errorf("hub status requires a matching ID and an expiry")
        }

diff --git a/protocol/message_test.go b/protocol/message_test.go
index 770ae501455b6364d83b8c82e88cb451c18982d7..79871dde99ee1b2f35164bd9d5ca49fb064aae2f 100644 (file)
--- a/protocol/message_test.go
+++ b/protocol/message_test.go
@@ -154,7 +154,7 @@ func TestPayloadValidationRejectsInvalidContracts(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
 
        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)
 
                        if tc.kind == Event {
                                require.Error(t, err)
 


@@ -167,7 +167,7 @@ func TestPayloadValidationRejectsInvalidContracts(t *testing.T) {
                                Kind:   tc.kind,
                                ID:     deviceID,
                                Target: tc.target,
                                Kind:   tc.kind,
                                ID:     deviceID,
                                Target: tc.target,

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

                        require.NoError(t, err)
 
                        _, err = Parse(topic, data, false)
                        require.NoError(t, err)
 
                        _, err = Parse(topic, data, false)

diff --git a/protocol/timing.go b/protocol/timing.go
new file mode 100644 (file)
index 0000000..e7fd410
--- /dev/null
+++ b/protocol/timing.go
@@ -0,0 +1,14 @@
+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
+)

diff --git a/protocol/envelope.go b/protocol/wire_envelope.go
similarity index 70%
rename from protocol/envelope.go
rename to protocol/wire_envelope.go
index a6833a4ca2f5ab6489bf9a70d1ba1e503020a52d..5e8b778e4613508b4945f4189dc8c9eaf6e3db8f 100644 (file)
--- a/protocol/envelope.go
+++ b/protocol/wire_envelope.go
@@ -9,19 +9,8 @@ import (
        "time"
 )
 
        "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.
 
 // Metadata holds message identity and optional UTC times. Retries keep the same
 // ID. A receiver checks ExpiresAt against its clock before processing.


@@ -33,21 +22,21 @@ type Metadata struct {
        CausationID   string     `json:"causation_id,omitempty"`
 }
 
        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"`
 }
 
        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)
        }
 
        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,
                Schema:   Schema,
                Metadata: metadata,
                Payload:  encoded,


@@ -65,25 +54,25 @@ func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) {
        return encoded, nil
 }
 
        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.
 // 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 {
        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 {
        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)
        }
        if envelope.Schema != Schema {
                return fmt.Errorf("unsupported schema %d", envelope.Schema)
        }

diff --git a/protocol/envelope_test.go b/protocol/wire_envelope_test.go
similarity index 87%
rename from protocol/envelope_test.go
rename to protocol/wire_envelope_test.go
index 4b55451fbffdf165bd37d5c7adb66b11b567a041..ce11f9107e15fe001b48656eeae6837ae638989d 100644 (file)
--- a/protocol/envelope_test.go
+++ b/protocol/wire_envelope_test.go
@@ -17,10 +17,10 @@ func TestEnvelopeRoundTripPreservesMetadata(t *testing.T) {
                CorrelationID: commandID,
                CausationID:   deviceID,
        }
                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)
 
        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)
        require.NoError(t, err)
        require.Equal(t, Schema, envelope.Schema)
        require.Equal(t, metadata, envelope.Metadata)


@@ -85,17 +85,17 @@ func TestEnvelopeRejectsInvalidWireData(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
 
        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)
                })
        }
 
                        require.Error(t, err)
                })
        }
 

-       _, err := EncodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{})
+       _, err := encodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{})

        require.Error(t, err)
 
        require.Error(t, err)
 

-       _, err = EncodeEnvelope(Metadata{MessageID: messageID}, make(chan int))
+       _, err = encodeEnvelope(Metadata{MessageID: messageID}, make(chan int))

        require.Error(t, err)
 
        require.Error(t, err)
 

-       _, err = EncodeEnvelope(Metadata{MessageID: messageID}, []int{1})
+       _, err = encodeEnvelope(Metadata{MessageID: messageID}, []int{1})

        require.Error(t, err)
 }
        require.Error(t, err)
 }