Tests use fake drivers. They do not verify a broker or a physical
device.
+
+******************
+ Typed publishing
+******************
+
+``Publisher`` takes a ``Transport`` with one ``Publish`` operation on
+``protocol.Publication``. No broker client or topic operation belongs in
+the framework. Typed methods cover descriptor, availability, reported
+values, telemetry, events, and results. Missing message IDs and
+observation times use fresh values. Supplied metadata remains subject to
+protocol validation.
+
+Equal retained payloads are suppressed. Correlation and causation bypass
+suppression. A changed telemetry sample uses QoS 0 without retention.
+Its latest complete value supplies a QoS 1 retained reported snapshot at
+the five-minute repeat or reconnect, not on every change. Repeats keep
+observation times, use fresh IDs, and omit command correlation and
+causation. JSON object key order and whitespace do not change equality.
+Measurement topics are independent. Boolean reported state does not
+repeat unless telemetry marks that point as a measurement. Online
+availability repeats every five minutes. Offline availability,
+descriptors, and Boolean state repeat on reconnect only. Expiring
+values, events, and results are not suppressed or cached.
+
+Run one publisher alongside the integration drivers. Its methods are
+safe for concurrent use and copy caller-owned payloads and timestamp
+pointers. Each send has a five-second timeout derived from the caller
+context. A background send error returns to the runner rather than being
+logged by the publisher. Reconnect refresh requests are nonblocking and
+coalesce while pending. The cache is memory-only. Tests use a fake
+transport and simulated time. They do not verify MQTT acknowledgement or
+a broker reconnect.
--- /dev/null
+package driver
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "time"
+
+ "onehouse/protocol"
+)
+
+type publicationValue struct {
+ publication protocol.Publication
+ address protocol.Address
+ metadata protocol.Metadata
+ payload any
+ comparison []byte
+ measurement bool
+ online bool
+}
+
+func newPublicationValue[T any](address protocol.Address, metadata protocol.Metadata, payload T) (publicationValue, error) {
+ metadata, err := newMetadata(metadata)
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("create publication metadata: %w", err)
+ }
+
+ value := publicationValue{
+ address: address,
+ metadata: metadata,
+ payload: payload,
+ }
+ value.publication, err = value.build()
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("build typed publication: %w", err)
+ }
+
+ data, err := json.Marshal(payload)
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("encode comparison payload: %w", err)
+ }
+
+ var normalized any
+
+ decoder := json.NewDecoder(bytes.NewReader(data))
+ decoder.UseNumber()
+
+ err = decoder.Decode(&normalized)
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("decode comparison payload: %w", err)
+ }
+
+ comparison, err := json.Marshal(normalized)
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("normalize comparison payload: %w", err)
+ }
+
+ var copied T
+
+ err = json.Unmarshal(comparison, &copied)
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("copy publication payload: %w", err)
+ }
+
+ value.payload = copied
+ value.comparison = comparison
+
+ switch typed := any(copied).(type) {
+ case protocol.ValuePayload:
+ value.measurement = address.Kind == protocol.Telemetry || string(typed.Value) != "true" && string(typed.Value) != "false"
+ case protocol.AvailabilityPayload:
+ value.online = typed.Status == "online"
+ }
+
+ return value, nil
+}
+
+func newMetadata(metadata protocol.Metadata) (protocol.Metadata, error) {
+ now := time.Now().UTC()
+ if metadata.MessageID == "" {
+ id, err := protocol.NewID(now)
+ if err != nil {
+ return protocol.Metadata{}, fmt.Errorf("create message ID: %w", err)
+ }
+
+ metadata.MessageID = id
+ }
+
+ if metadata.SentAt == nil {
+ metadata.SentAt = &now
+ } else {
+ timestamp := *metadata.SentAt
+ metadata.SentAt = ×tamp
+ }
+
+ if metadata.ExpiresAt != nil {
+ timestamp := *metadata.ExpiresAt
+ metadata.ExpiresAt = ×tamp
+ }
+
+ return metadata, nil
+}
+
+func (v publicationValue) build() (protocol.Publication, error) {
+ var (
+ publication protocol.Publication
+ err error
+ )
+
+ switch payload := v.payload.(type) {
+ case protocol.DescriptorPayload:
+ publication, err = protocol.BuildDescriptor(v.address.ID, v.metadata, payload)
+ case protocol.AvailabilityPayload:
+ publication, err = protocol.BuildAvailability(v.address.ID, v.metadata, payload)
+ case protocol.ValuePayload:
+ if v.address.Kind == protocol.Telemetry {
+ publication, err = protocol.BuildTelemetry(v.address.ID, v.address.Target, v.metadata, payload)
+ } else {
+ publication, err = protocol.BuildReported(v.address.ID, v.address.Target, v.metadata, payload)
+ }
+ case protocol.EventPayload:
+ publication, err = protocol.BuildEvent(v.address.ID, v.address.Target, v.metadata, payload)
+ case protocol.ResultPayload:
+ publication, err = protocol.BuildResult(v.address.ID, v.address.Target, v.metadata, payload)
+ default:
+ return protocol.Publication{}, fmt.Errorf("unsupported publication payload")
+ }
+
+ if err != nil {
+ return protocol.Publication{}, fmt.Errorf("build %s: %w", v.address.Kind, err)
+ }
+
+ return publication, nil
+}
+
+func (v publicationValue) repeated() (publicationValue, error) {
+ metadata, err := newMetadata(protocol.Metadata{SentAt: v.metadata.SentAt})
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("create repeat metadata: %w", err)
+ }
+
+ v.metadata = metadata
+ if v.address.Kind == protocol.Telemetry {
+ v.address.Kind = protocol.Reported
+ }
+
+ v.publication, err = v.build()
+ if err != nil {
+ return publicationValue{}, fmt.Errorf("build repeated publication: %w", err)
+ }
+
+ return v, nil
+}
--- /dev/null
+package driver
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "sync"
+ "time"
+
+ "onehouse/protocol"
+)
+
+const publishTimeout = 5 * time.Second
+
+// Publisher sends typed device messages and owns retained-value suppression.
+// Its methods are safe for concurrent use. Run owns repeats and Refresh
+// requests reconnect snapshots. The caller supplies a valid transport and runs
+// this publisher once alongside the integration drivers.
+type Publisher struct {
+ transport Transport
+ mu sync.Mutex
+ retained map[protocol.Address]*retainedValue
+ refresh chan struct{}
+}
+
+// NewPublisher creates a publisher with no cached state. The caller retains
+// ownership of transport and must keep it available until Run stops.
+func NewPublisher(transport Transport) *Publisher {
+ return &Publisher{
+ transport: transport,
+ retained: make(map[protocol.Address]*retainedValue),
+ refresh: make(chan struct{}, 1),
+ }
+}
+
+// Run repeats measurements and online evidence at the contract interval. A send
+// failure returns to the runner. Observation times stay unchanged and repeated
+// messages have fresh IDs without command correlation or causation.
+func (p *Publisher) Run(ctx context.Context) error {
+ ticker := time.NewTicker(250 * time.Millisecond)
+ defer ticker.Stop()
+
+ for {
+ force := false
+
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-p.refresh:
+ force = true
+ case <-ticker.C:
+ }
+
+ err := p.repeat(ctx, force)
+ if err != nil {
+ return fmt.Errorf("repeat snapshots: %w", err)
+ }
+ }
+}
+
+// Refresh coalesces a nonblocking request to resend all cached snapshots.
+func (p *Publisher) Refresh() {
+ select {
+ case p.refresh <- struct{}{}:
+ default:
+ }
+}
+
+// Descriptor sends retained metadata unless its value equals the last success.
+// Missing message ID and observation time are filled by the publisher. A
+// supplied correlation or causation always bypasses value suppression.
+func (p *Publisher) Descriptor(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.DescriptorPayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Descriptor,
+ ID: deviceID,
+ }, metadata, payload)
+}
+
+// Availability sends reachability and repeats online evidence during Run.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Availability(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.AvailabilityPayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Availability,
+ ID: deviceID,
+ }, metadata, payload)
+}
+
+// Reported sends a complete retained value. Non-Boolean values repeat during
+// Run. A Boolean point also repeats after Telemetry marks it as a measurement.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Reported(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Reported,
+ ID: deviceID,
+ Target: capability,
+ }, metadata, payload)
+}
+
+// Telemetry sends changed measurements at QoS 0 without retention. The latest
+// observation supplies the retained reported snapshot on repeat or reconnect,
+// not on every telemetry change. Equal samples are suppressed until repeat.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Telemetry(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Telemetry,
+ ID: deviceID,
+ Target: capability,
+ }, metadata, payload)
+}
+
+// Event sends each occurrence without suppression. Metadata defaults follow
+// Descriptor. The caller owns event-specific observation and safety rules.
+func (p *Publisher) Event(ctx context.Context, deviceID string, event string, metadata protocol.Metadata, payload protocol.EventPayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Event,
+ ID: deviceID,
+ Target: event,
+ }, metadata, payload)
+}
+
+// Result sends an outcome without suppression. Correlation must equal the
+// command ID. Reuse ledger metadata to replay the same result ID and time.
+// Missing message ID and observation time follow Descriptor defaults.
+func (p *Publisher) Result(ctx context.Context, deviceID string, commandID string, metadata protocol.Metadata, payload protocol.ResultPayload) error {
+ return publishTyped(ctx, p, protocol.Address{
+ Kind: protocol.Result,
+ ID: deviceID,
+ Target: commandID,
+ }, metadata, payload)
+}
+
+func (p *Publisher) publish(ctx context.Context, value publicationValue) error {
+ err := ctx.Err()
+ if err != nil {
+ return err
+ }
+
+ if value.metadata.ExpiresAt != nil && !time.Now().Before(*value.metadata.ExpiresAt) {
+ return fmt.Errorf("publication expired")
+ }
+
+ publication := value.publication
+
+ if !publication.Retain && value.address.Kind != protocol.Telemetry || value.metadata.ExpiresAt != nil {
+ err = p.send(ctx, publication)
+ if err != nil {
+ return fmt.Errorf("send direct publication: %w", err)
+ }
+
+ return nil
+ }
+
+ key := value.address
+ if key.Kind == protocol.Telemetry {
+ key.Kind = protocol.Reported
+ }
+
+ p.mu.Lock()
+
+ entry := p.retained[key]
+ if entry == nil {
+ entry = &retainedValue{retainedAt: time.Now()}
+ p.retained[key] = entry
+ }
+
+ p.mu.Unlock()
+ entry.mu.Lock()
+ defer entry.mu.Unlock()
+
+ entry.latest = value
+ entry.measurement = entry.measurement || value.measurement
+ now := time.Now()
+ equal := bytes.Equal(entry.successful, value.comparison)
+ correlated := value.metadata.CorrelationID != "" || value.metadata.CausationID != ""
+ due := entry.measurement && !now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || value.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
+ if equal && !correlated && !due {
+ return nil
+ }
+
+ if entry.measurement && equal && due && !correlated {
+ value, err = value.repeated()
+ if err != nil {
+ return fmt.Errorf("prepare reported repeat: %w", err)
+ }
+
+ publication = value.publication
+ }
+
+ err = p.send(ctx, publication)
+ if err != nil {
+ return fmt.Errorf("send publication: %w", err)
+ }
+
+ entry.successful = value.comparison
+ entry.publishedAt = now
+ if publication.Retain {
+ entry.retainedAt = now
+ }
+
+ return nil
+}
+
+func (p *Publisher) send(ctx context.Context, publication protocol.Publication) error {
+ err := ctx.Err()
+ if err != nil {
+ return err
+ }
+
+ if publication.ExpiresAt != nil && !time.Now().Before(*publication.ExpiresAt) {
+ return fmt.Errorf("publication expired")
+ }
+
+ ctx, cancel := context.WithTimeout(ctx, publishTimeout)
+ defer cancel()
+
+ err = p.transport.Publish(ctx, publication)
+ if err != nil {
+ return fmt.Errorf("send transport publication: %w", err)
+ }
+
+ return nil
+}
+
+func (p *Publisher) repeat(ctx context.Context, force bool) error {
+ p.mu.Lock()
+
+ entries := make([]*retainedValue, 0, len(p.retained))
+
+ for _, entry := range p.retained {
+ entries = append(entries, entry)
+ }
+
+ p.mu.Unlock()
+
+ for _, entry := range entries {
+ err := p.repeatEntry(ctx, entry, force)
+ if err != nil {
+ return fmt.Errorf("repeat retained value: %w", err)
+ }
+ }
+
+ return nil
+}
+
+func (p *Publisher) repeatEntry(ctx context.Context, entry *retainedValue, force bool) error {
+ entry.mu.Lock()
+ defer entry.mu.Unlock()
+
+ now := time.Now()
+ due := entry.measurement && (!now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || !bytes.Equal(entry.successful, entry.latest.comparison)) || entry.latest.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
+ if !force && !due {
+ return nil
+ }
+
+ value, err := entry.latest.repeated()
+ if err != nil {
+ return fmt.Errorf("prepare snapshot: %w", err)
+ }
+
+ err = p.send(ctx, value.publication)
+ if err != nil {
+ return fmt.Errorf("send snapshot: %w", err)
+ }
+
+ entry.successful = value.comparison
+ entry.publishedAt = now
+ entry.retainedAt = now
+
+ return nil
+}
+
+func publishTyped[T any](ctx context.Context, publisher *Publisher, address protocol.Address, metadata protocol.Metadata, payload T) error {
+ value, err := newPublicationValue(address, metadata, payload)
+ if err != nil {
+ return fmt.Errorf("prepare typed publication: %w", err)
+ }
+
+ err = publisher.publish(ctx, value)
+ if err != nil {
+ return fmt.Errorf("publish %s: %w", address.Kind, err)
+ }
+
+ return nil
+}
--- /dev/null
+package driver
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestPublisherSuppressesEqualRetainedValuesAndKeepsCommandEvidence(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ metadata := protocol.Metadata{}
+ descriptor := protocol.DescriptorPayload{
+ Manufacturer: "example",
+ Model: "switch",
+ Firmware: "1",
+ Capabilities: []protocol.Capability{{
+ ID: "relay",
+ Type: "switch",
+ ReportsState: true,
+ }},
+ }
+
+ for range 2 {
+ require.NoError(t, publisher.Descriptor(t.Context(), testDeviceID, metadata, descriptor))
+ require.NoError(t, publisher.Availability(t.Context(), testDeviceID, metadata, protocol.AvailabilityPayload{Status: "offline"}))
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", metadata, protocol.ValuePayload{Value: json.RawMessage("false")}))
+ }
+
+ require.Len(t, transport.messages(), 3)
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CausationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
+ require.Len(t, transport.messages(), 5)
+
+ for _, publication := range transport.messages() {
+ require.Equal(t, byte(1), publication.QoS)
+ require.True(t, publication.Retain)
+ }
+}
+
+func TestPublisherRepeatsLatestMeasurementWithoutMovingDeadline(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ finished := make(chan error, 1)
+
+ require.NoError(t, publisher.Reported(ctx, testDeviceID, "weather", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)}))
+
+ first := transport.messages()[0]
+ go func() {
+ finished <- publisher.Run(ctx)
+ }()
+
+ for range 299 {
+ time.Sleep(time.Second)
+ require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "weather", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage(`{ "humidity":50,"temperature":20 }`)}))
+ synctest.Wait()
+ require.Len(t, transport.messages(), 1)
+ }
+
+ time.Sleep(time.Second)
+ synctest.Wait()
+
+ packets := transport.messages()
+ require.Len(t, packets, 2)
+
+ repeated, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+ require.NoError(t, err)
+
+ original, err := protocol.ParseReported(first.Topic, first.Data, first.Retain)
+ require.NoError(t, err)
+ require.NotEqual(t, original.Metadata.MessageID, repeated.Metadata.MessageID)
+ require.Equal(t, time.Now().Add(-time.Second).UTC(), *repeated.Metadata.SentAt)
+ require.Empty(t, repeated.Metadata.CorrelationID)
+ require.Empty(t, repeated.Metadata.CausationID)
+ require.JSONEq(t, `{"temperature":20,"humidity":50}`, string(repeated.Payload.Value))
+ cancel()
+ require.ErrorIs(t, <-finished, context.Canceled)
+ })
+}
+
+func TestPublisherTelemetryKeepsSnapshotOnScheduleAndReconnect(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ finished := make(chan error, 1)
+
+ require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("10")}))
+
+ go func() {
+ finished <- publisher.Run(ctx)
+ }()
+
+ time.Sleep(4 * time.Minute)
+ require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
+ require.NoError(t, publisher.Telemetry(ctx, testOtherDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("30")}))
+ synctest.Wait()
+
+ packets := transport.messages()
+ require.Len(t, packets, 3)
+
+ for _, packet := range packets {
+ require.Equal(t, byte(0), packet.QoS)
+ require.False(t, packet.Retain)
+ }
+
+ time.Sleep(time.Minute)
+ synctest.Wait()
+
+ packets = transport.messages()
+ require.Len(t, packets, 4)
+
+ snapshot, err := protocol.ParseReported(packets[3].Topic, packets[3].Data, packets[3].Retain)
+ require.NoError(t, err)
+ require.JSONEq(t, "20", string(snapshot.Payload.Value))
+ require.Equal(t, byte(1), packets[3].QoS)
+ require.Equal(t, time.Now().Add(-time.Minute).UTC(), *snapshot.Metadata.SentAt)
+ publisher.Refresh()
+ synctest.Wait()
+ require.Len(t, transport.messages(), 6)
+ cancel()
+ require.ErrorIs(t, <-finished, context.Canceled)
+ })
+}
+
+func TestPublisherRepeatsOnlineButNotBooleanOrOffline(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ finished := make(chan error, 1)
+
+ require.NoError(t, publisher.Availability(ctx, testDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"}))
+ require.NoError(t, publisher.Availability(ctx, testOtherDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"}))
+ require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+
+ go func() {
+ finished <- publisher.Run(ctx)
+ }()
+
+ time.Sleep(5 * time.Minute)
+ synctest.Wait()
+
+ packets := transport.messages()
+ require.Len(t, packets, 4)
+
+ online, err := protocol.ParseAvailability(packets[3].Topic, packets[3].Data, packets[3].Retain)
+ require.NoError(t, err)
+ require.Equal(t, "online", online.Payload.Status)
+ publisher.Refresh()
+ synctest.Wait()
+ require.Len(t, transport.messages(), 7)
+ cancel()
+ require.ErrorIs(t, <-finished, context.Canceled)
+ })
+}
+
+func TestPublisherDoesNotSuppressEventsResultsOrExpiringValues(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ expiry := time.Now().UTC().Add(time.Minute)
+
+ for range 2 {
+ require.NoError(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{"action": json.RawMessage(`"pressed"`)}))
+ require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, protocol.Metadata{CorrelationID: testCommandID}, protocol.ResultPayload{Status: "completed"}))
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{ExpiresAt: &expiry}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+ }
+
+ require.Len(t, transport.messages(), 6)
+ require.NoError(t, publisher.repeat(t.Context(), true))
+ require.Len(t, transport.messages(), 6)
+}
+
+func TestPublisherKeepsFailedLatestSampleAndReturnsRepeatFailures(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
+
+ failure := errors.New("broker unavailable")
+ transport.setError(failure)
+ require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("21")}), failure)
+ require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("22")}), failure)
+ publisher.Refresh()
+ require.ErrorIs(t, publisher.Run(t.Context()), failure)
+ transport.setError(nil)
+ require.NoError(t, publisher.repeat(t.Context(), false))
+
+ packets := transport.messages()
+ last := packets[len(packets)-1]
+ snapshot, err := protocol.ParseReported(last.Topic, last.Data, last.Retain)
+ require.NoError(t, err)
+ require.JSONEq(t, "22", string(snapshot.Payload.Value))
+ })
+}
+
+func TestPublisherValidatesBeforeSendingAndHonorsCancellation(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ require.Error(t, publisher.Reported(t.Context(), "invalid", "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+ require.Error(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("invalid")}))
+
+ past := time.Now().UTC().Add(-time.Second)
+ require.ErrorContains(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{ExpiresAt: &past}, protocol.EventPayload{}), "expired")
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ cancel()
+ require.ErrorIs(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}), context.Canceled)
+ require.Empty(t, transport.messages())
+}
+
+func TestPublisherSendTimeoutAndCallerCancellation(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ publisher := NewPublisher(&waitingTransport{})
+ err := publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
+ require.ErrorIs(t, err, context.DeadlineExceeded)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ finished := make(chan error, 1)
+ go func() {
+ finished <- publisher.Event(ctx, testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
+ }()
+
+ synctest.Wait()
+ cancel()
+ require.ErrorIs(t, <-finished, context.Canceled)
+ })
+}
+
+func TestPublisherCopiesSnapshotsAndAcceptsTypedNull(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ now := time.Now().UTC()
+ metadata := protocol.Metadata{SentAt: &now}
+ value := json.RawMessage(`{"power":10}`)
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{Value: value}))
+
+ observation := now
+ now = now.Add(time.Hour)
+
+ copy(value, `{"power":99}`)
+ publisher.Refresh()
+ publisher.Refresh()
+ require.NoError(t, publisher.repeat(t.Context(), true))
+
+ packets := transport.messages()
+ require.Len(t, packets, 2)
+
+ snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+ require.NoError(t, err)
+ require.JSONEq(t, `{"power":10}`, string(snapshot.Payload.Value))
+ require.Equal(t, observation, *snapshot.Metadata.SentAt)
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{}))
+}
+
+func TestPublisherEqualTelemetryAtRepeatTimeSendsRetainedSnapshot(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ value := protocol.ValuePayload{Value: json.RawMessage("null")}
+ require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
+ time.Sleep(protocol.MeasurementRepeatInterval)
+ require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
+
+ packets := transport.messages()
+ require.Len(t, packets, 2)
+ require.True(t, packets[1].Retain)
+
+ snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+ require.NoError(t, err)
+ require.JSONEq(t, "null", string(snapshot.Payload.Value))
+ })
+}
+
+func TestPublisherPreservesDirectMeasurementEvidenceAtRepeatDeadline(t *testing.T) {
+ testCases := []struct {
+ name string
+ correlationID string
+ causationID string
+ }{
+ {
+ name: "correlation keeps caller identity",
+ correlationID: testCommandID,
+ },
+ {
+ name: "causation keeps caller identity",
+ causationID: testCommandID,
+ },
+ {
+ name: "correlation and causation keep caller identity",
+ correlationID: testCommandID,
+ causationID: testOtherDeviceID,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ value := protocol.ValuePayload{Value: json.RawMessage("20")}
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, value))
+ time.Sleep(protocol.MeasurementRepeatInterval)
+
+ observedAt := time.Now().UTC()
+ metadata := protocol.Metadata{
+ MessageID: testOtherDeviceID,
+ SentAt: &observedAt,
+ CorrelationID: tc.correlationID,
+ CausationID: tc.causationID,
+ }
+ require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", metadata, value))
+
+ packets := transport.messages()
+ require.Len(t, packets, 2)
+
+ message, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+ require.NoError(t, err)
+ require.Equal(t, metadata, message.Metadata)
+ require.Equal(t, value, message.Payload)
+ })
+ })
+ }
+}
+
+type waitingTransport struct{}
+
+func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
+ <-ctx.Done()
+
+ return ctx.Err()
+}
--- /dev/null
+package driver
+
+import (
+ "sync"
+ "time"
+)
+
+type retainedValue struct {
+ mu sync.Mutex
+ latest publicationValue
+ successful []byte
+ publishedAt time.Time
+ retainedAt time.Time
+ measurement bool
+}
for index, integration := range drivers {
workers.Go(func() {
- results <- driverResult{index: index, err: integration.Run(ctx)}
+ results <- driverResult{
+ index: index,
+ err: integration.Run(ctx),
+ }
})
}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
- first := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
- second := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
+ first := &fakeDriver{
+ stopped: make(chan struct{}),
+ refreshed: make(chan struct{}, 1),
+ }
+ second := &fakeDriver{
+ stopped: make(chan struct{}),
+ refreshed: make(chan struct{}, 1),
+ }
connected := make(chan struct{}, 1)
finished := make(chan error, 1)
go func() {
canceled bool
failure bool
}{
- {name: "normal completion", err: nil},
- {name: "cancellation is a stop", err: context.Canceled, canceled: true},
- {name: "closed transport is a stop", err: net.ErrClosed, canceled: true},
- {name: "closed pipe is a stop", err: io.ErrClosedPipe, canceled: true},
- {name: "broken pipe is a stop", err: syscall.EPIPE, canceled: true},
- {name: "reset connection is a stop", err: syscall.ECONNRESET, canceled: true},
- {name: "wrapped stop stays a stop", err: fmt.Errorf("read: %w", net.ErrClosed), canceled: true},
- {name: "joined stops stay stops", err: errors.Join(context.Canceled, net.ErrClosed), canceled: true},
- {name: "joined failure stays a failure", err: errors.Join(context.Canceled, failure), canceled: true, failure: true},
- {name: "request timeout stays a failure", err: context.DeadlineExceeded, failure: true},
- {name: "active transport close is a failure", err: net.ErrClosed, failure: true},
- {name: "independent failure stays a failure", err: failure, canceled: true, failure: true},
+ {
+ name: "normal completion",
+ err: nil,
+ },
+ {
+ name: "cancellation is a stop",
+ err: context.Canceled,
+ canceled: true,
+ },
+ {
+ name: "closed transport is a stop",
+ err: net.ErrClosed,
+ canceled: true,
+ },
+ {
+ name: "closed pipe is a stop",
+ err: io.ErrClosedPipe,
+ canceled: true,
+ },
+ {
+ name: "broken pipe is a stop",
+ err: syscall.EPIPE,
+ canceled: true,
+ },
+ {
+ name: "reset connection is a stop",
+ err: syscall.ECONNRESET,
+ canceled: true,
+ },
+ {
+ name: "wrapped stop stays a stop",
+ err: fmt.Errorf("read: %w", net.ErrClosed),
+ canceled: true,
+ },
+ {
+ name: "joined stops stay stops",
+ err: errors.Join(context.Canceled, net.ErrClosed),
+ canceled: true,
+ },
+ {
+ name: "joined failure stays a failure",
+ err: errors.Join(context.Canceled, failure),
+ canceled: true,
+ failure: true,
+ },
+ {
+ name: "request timeout stays a failure",
+ err: context.DeadlineExceeded,
+ failure: true,
+ },
+ {
+ name: "active transport close is a failure",
+ err: net.ErrClosed,
+ failure: true,
+ },
+ {
+ name: "independent failure stays a failure",
+ err: failure,
+ canceled: true,
+ failure: true,
+ },
}
for _, tc := range testCases {
--- /dev/null
+package driver
+
+import (
+ "context"
+
+ "onehouse/protocol"
+)
+
+// Transport sends a complete protocol publication. The caller owns the
+// connection. Publish must stop when ctx stops and must not mutate publication.
+type Transport interface {
+ Publish(ctx context.Context, publication protocol.Publication) error
+}
--- /dev/null
+package driver
+
+import (
+ "context"
+ "sync"
+
+ "onehouse/protocol"
+)
+
+const (
+ testDeviceID = "01K2K7M5TZ3J7W4Q9X8N6P2RVA"
+ testOtherDeviceID = "01K2K7M5TZ3J7W4Q9X8N6P2RVB"
+ testCommandID = "01K2K7M5TZ3J7W4Q9X8N6P2RVF"
+)
+
+type recordingTransport struct {
+ mu sync.Mutex
+ publications []protocol.Publication
+ err error
+}
+
+func (p *recordingTransport) Publish(_ context.Context, publication protocol.Publication) error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ p.publications = append(p.publications, publication)
+
+ return p.err
+}
+
+func (p *recordingTransport) messages() []protocol.Publication {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ return append([]protocol.Publication(nil), p.publications...)
+}
+
+func (p *recordingTransport) setError(err error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ p.err = err
+}
projects / onehouse / commitdiff