##########
OneHouse is one program that controls a whole house. Linux is supported.
+The command runs an empty driver list and stops on SIGINT or SIGTERM.
+``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines
+its wire contract.
*************
Development
nix flake check
``AGENTS.md`` holds the rules for work in this repository.
+
+Run the command with ``nix run``. It waits for a signal without starting
+an integration or a broker.
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.
+
+*******************
+ Command ownership
+*******************
+
+``NewRouter`` takes the explicit IDs of command-capable devices. It
+supplies protocol subscriptions and one 64-entry FIFO per device. Each
+entry holds one single command or one ordered batch. Intake parses a
+delivery once through ``protocol.Parse``. A malformed or expired member
+rejects the whole batch. Retained commands and commands for unowned
+devices are errors. A full queue waits for space, cancellation, or the
+earliest member expiry. Each device has one queue consumer. It checks
+physical ownership, capability access, and expiry again before work.
+Read-only devices have no queue.
+
+``NewLedger`` creates a ledger for one device. The consumer checks
+``Lookup`` before a physical action, and stores its result with
+``Remember`` before the first publication attempt. A duplicate reuses
+the first status, reason, message ID, and observation time. It performs
+no second action. This also applies after a failed result send. Pass the
+returned ``ResultRecord`` fields to ``Publisher.Result`` unchanged.
+Records are safe from caller mutation.
+
+The ledger retains 256 results in insertion order per device. Lookup
+does not extend their age. It is not a reservation for concurrent
+physical actions. The single queue consumer provides that exclusion. The
+ledger is memory-only. Eviction or process restart removes duplicate
+protection for that record. Pending physical confirmation and
+device-specific command safety belong to the integration. Fake tests
+verify queue order, backpressure, expiry, per-device bounds, and stable
+duplicate replay without device I/O.
--- /dev/null
+package driver
+
+import (
+ "time"
+
+ "onehouse/protocol"
+)
+
+// Command is one validated device request. A driver checks ownership, access,
+// and Expired immediately before physical work. Each batch member keeps its own
+// metadata, expiry, and payload. The queue consumer owns these values.
+type Command struct {
+ DeviceID string
+ Capability string
+ Metadata protocol.Metadata
+ Payload protocol.ValuePayload
+}
+
+// Expired reports whether the command has no expiry or has reached it.
+func (c Command) Expired(now time.Time) bool {
+ return c.Metadata.ExpiresAt == nil || !now.Before(*c.Metadata.ExpiresAt)
+}
--- /dev/null
+package driver
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestCommandExpiryIncludesBoundaryAndMissingDeadline(t *testing.T) {
+ now := time.Now().UTC()
+ expiry := now.Add(time.Minute)
+ command := Command{Metadata: protocol.Metadata{ExpiresAt: &expiry}}
+ require.False(t, command.Expired(now))
+ require.True(t, command.Expired(expiry))
+ require.True(t, Command{}.Expired(now))
+}
--- /dev/null
+package driver
+
+type driverResult struct {
+ index int
+ err error
+}
--- /dev/null
+package driver
+
+import (
+ "fmt"
+ "sync"
+
+ "onehouse/protocol"
+)
+
+const commandResultCapacity = 256
+
+// Ledger keeps the first result of each of the most recent 256 commands for one
+// device, in insertion order. Its methods are safe for concurrent use. A single
+// queue consumer checks Lookup before acting and calls Remember before sending
+// the result, also after a failed send. The ledger does not reserve physical
+// work, persist records, or prevent replay after eviction.
+type Ledger struct {
+ deviceID string
+ mu sync.Mutex
+ results map[string]ResultRecord
+ order []string
+}
+
+// NewLedger creates an empty result ledger for one canonical device ID.
+func NewLedger(deviceID string) (*Ledger, error) {
+ if !protocol.ValidID(deviceID) {
+ return nil, fmt.Errorf("ledger device ID must be a canonical ULID")
+ }
+
+ return &Ledger{
+ deviceID: deviceID,
+ results: make(map[string]ResultRecord),
+ order: make([]string, 0, commandResultCapacity),
+ }, nil
+}
+
+// Lookup returns an independent copy of a result without changing its age.
+func (l *Ledger) Lookup(commandID string) (ResultRecord, bool) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+
+ result, exists := l.results[commandID]
+
+ return result.clone(), exists
+}
+
+// Remember stores the first validated result before publication. An existing
+// result wins even when a duplicate proposes another status or reason. Its ID
+// and observation time remain stable after failed publication and on replay.
+func (l *Ledger) Remember(commandID string, payload protocol.ResultPayload) (ResultRecord, error) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+
+ result, exists := l.results[commandID]
+ if exists {
+ return result.clone(), nil
+ }
+
+ metadata, err := newMetadata(protocol.Metadata{CorrelationID: commandID})
+ if err != nil {
+ return ResultRecord{}, fmt.Errorf("create result metadata: %w", err)
+ }
+
+ _, err = protocol.BuildResult(l.deviceID, commandID, metadata, payload)
+ if err != nil {
+ return ResultRecord{}, fmt.Errorf("validate result: %w", err)
+ }
+
+ if len(l.order) == commandResultCapacity {
+ delete(l.results, l.order[0])
+
+ l.order = l.order[1:]
+ }
+
+ result = ResultRecord{
+ Metadata: metadata,
+ Payload: payload,
+ }
+ l.results[commandID] = result
+ l.order = append(l.order, commandID)
+
+ return result.clone(), nil
+}
--- /dev/null
+package driver
+
+import (
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestLedgerReplaysFirstResultWithoutAnotherAction(t *testing.T) {
+ ledger, err := NewLedger(testDeviceID)
+ require.NoError(t, err)
+
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ actions := 0
+
+ var first ResultRecord
+
+ for attempt := range 3 {
+ result, exists := ledger.Lookup(testCommandID)
+ if !exists {
+ actions++
+ result, err = ledger.Remember(testCommandID, protocol.ResultPayload{Status: "completed"})
+ require.NoError(t, err)
+
+ first = result
+ }
+
+ if attempt == 0 {
+ transport.setError(errors.New("delivery uncertain"))
+ require.Error(t, publisher.Result(t.Context(), testDeviceID, testCommandID, result.Metadata, result.Payload))
+ transport.setError(nil)
+ } else {
+ require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, result.Metadata, result.Payload))
+ }
+
+ require.Equal(t, first, result)
+ }
+
+ require.Equal(t, 1, actions)
+
+ packets := transport.messages()
+ require.Len(t, packets, 3)
+ require.Equal(t, packets[0], packets[1])
+ require.Equal(t, packets[0], packets[2])
+
+ ignored, err := ledger.Remember(testCommandID, protocol.ResultPayload{
+ Status: "failed",
+ Reason: "different outcome",
+ })
+ require.NoError(t, err)
+ require.Equal(t, first, ignored)
+}
+
+func TestLedgerEvictsOldestAndKeepsDevicesIndependent(t *testing.T) {
+ ledger, err := NewLedger(testDeviceID)
+ require.NoError(t, err)
+
+ first, err := ledger.Remember(testCommandID, protocol.ResultPayload{
+ Status: "rejected",
+ Reason: "invalid output",
+ })
+ require.NoError(t, err)
+
+ other, err := NewLedger(testOtherDeviceID)
+ require.NoError(t, err)
+
+ independent, err := other.Remember(testCommandID, protocol.ResultPayload{
+ Status: "failed",
+ Reason: "unavailable",
+ })
+ require.NoError(t, err)
+ require.NotEqual(t, first.Payload, independent.Payload)
+
+ lastID := ""
+
+ for range commandResultCapacity {
+ lastID, err = protocol.NewID(time.Now().UTC())
+ require.NoError(t, err)
+
+ _, err = ledger.Remember(lastID, protocol.ResultPayload{Status: "completed"})
+ require.NoError(t, err)
+ }
+
+ _, exists := ledger.Lookup(testCommandID)
+ require.False(t, exists)
+
+ _, exists = ledger.Lookup(lastID)
+ require.True(t, exists)
+
+ _, exists = other.Lookup(testCommandID)
+ require.True(t, exists)
+}
+
+func TestLedgerRejectsInvalidResultsAndCopiesReturnedTimes(t *testing.T) {
+ _, err := NewLedger("invalid")
+ require.Error(t, err)
+
+ ledger, err := NewLedger(testDeviceID)
+ require.NoError(t, err)
+
+ testCases := []struct {
+ name string
+ payload protocol.ResultPayload
+ }{
+ {
+ name: "failure requires a reason",
+ payload: protocol.ResultPayload{Status: "failed"},
+ },
+ {
+ name: "completion rejects a reason",
+ payload: protocol.ResultPayload{
+ Status: "completed",
+ Reason: "not allowed",
+ },
+ },
+ {
+ name: "unknown status rejects",
+ payload: protocol.ResultPayload{Status: "invalid"},
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ _, err = ledger.Remember(testCommandID, tc.payload)
+ require.Error(t, err)
+
+ _, exists := ledger.Lookup(testCommandID)
+ require.False(t, exists)
+ })
+ }
+
+ _, err = ledger.Remember("invalid", protocol.ResultPayload{Status: "completed"})
+ require.Error(t, err)
+
+ result, err := ledger.Remember(testCommandID, protocol.ResultPayload{Status: "completed"})
+ require.NoError(t, err)
+
+ observation := *result.Metadata.SentAt
+ *result.Metadata.SentAt = observation.Add(time.Hour)
+ first, exists := ledger.Lookup(testCommandID)
+ require.True(t, exists)
+ require.Equal(t, observation, *first.Metadata.SentAt)
+
+ *first.Metadata.SentAt = observation.Add(time.Hour)
+ second, err := ledger.Remember(testCommandID, protocol.ResultPayload{})
+ require.NoError(t, err)
+ require.Equal(t, observation, *second.Metadata.SentAt)
+}
}
}
+func TestPublisherRunsAsDriverAndRefreshesOnReconnect(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ transport := &recordingTransport{}
+ publisher := NewPublisher(transport)
+ require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+
+ connected := make(chan struct{}, 1)
+ finished := make(chan error, 1)
+ go func() {
+ finished <- Run(ctx, []Driver{publisher}, connected)
+ }()
+
+ synctest.Wait()
+
+ connected <- struct{}{}
+
+ synctest.Wait()
+
+ packets := transport.messages()
+ require.Len(t, packets, 2)
+ require.NotEqual(t, packets[0].Data, packets[1].Data)
+ cancel()
+ require.NoError(t, <-finished)
+ })
+}
+
type waitingTransport struct{}
func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
--- /dev/null
+package driver
+
+import "onehouse/protocol"
+
+// ResultRecord holds the first result and its stable replay metadata. Pass both
+// fields unchanged to Publisher.Result. Lookup and Remember return independent
+// timestamp pointers, so callers cannot change the ledger.
+type ResultRecord struct {
+ Metadata protocol.Metadata
+ Payload protocol.ResultPayload
+}
+
+func (r ResultRecord) clone() ResultRecord {
+ if r.Metadata.SentAt != nil {
+ timestamp := *r.Metadata.SentAt
+ r.Metadata.SentAt = ×tamp
+ }
+
+ return r
+}
--- /dev/null
+package driver
+
+import (
+ "context"
+ "fmt"
+ "slices"
+ "time"
+
+ "onehouse/protocol"
+)
+
+const commandQueueCapacity = 64
+
+// Router owns one bounded FIFO for each supplied command-capable device. A
+// queue entry is one single command or one ordered batch. Handle is safe for
+// concurrent calls. Each queue has one consumer that serializes device work.
+// The caller stops producers and consumers through their contexts. Queues do
+// not close and belong to the router for its lifetime.
+type Router struct {
+ queues map[string]chan []Command
+ subscriptions []protocol.Subscription
+}
+
+// NewRouter validates unique device IDs and makes 64-entry queues. Read-only
+// devices do not belong in deviceIDs. The constructor performs no I/O.
+func NewRouter(deviceIDs []string) (*Router, error) {
+ ids := slices.Clone(deviceIDs)
+ slices.Sort(ids)
+
+ router := &Router{queues: make(map[string]chan []Command, len(deviceIDs))}
+
+ for _, deviceID := range ids {
+ _, exists := router.queues[deviceID]
+ if exists {
+ return nil, fmt.Errorf("duplicate command device ID")
+ }
+
+ subscription, err := protocol.CommandSubscription(deviceID)
+ if err != nil {
+ return nil, fmt.Errorf("build command subscription: %w", err)
+ }
+
+ router.queues[deviceID] = make(chan []Command, commandQueueCapacity)
+ router.subscriptions = append(router.subscriptions, subscription)
+ }
+
+ return router, nil
+}
+
+// Subscriptions returns an independent, sorted list including batch intake.
+func (r *Router) Subscriptions() []protocol.Subscription {
+ return slices.Clone(r.subscriptions)
+}
+
+// Queue returns the receive-only command queue for a configured device.
+func (r *Router) Queue(deviceID string) (<-chan []Command, bool) {
+ queue, exists := r.queues[deviceID]
+
+ return queue, exists
+}
+
+// Handle parses one delivery once and queues all its commands together. It
+// rejects malformed, retained, expired, non-command, and unowned deliveries. A
+// full queue waits until space, caller cancellation, or the earliest member
+// expiry. A driver must check expiry again when it consumes the queue entry.
+func (r *Router) Handle(ctx context.Context, topic string, data []byte, retained bool) error {
+ err := ctx.Err()
+ if err != nil {
+ return err
+ }
+
+ message, err := protocol.Parse(topic, data, retained)
+ if err != nil {
+ return fmt.Errorf("parse command delivery: %w", err)
+ }
+
+ var commands []Command
+
+ switch payload := message.Payload.(type) {
+ case protocol.ValuePayload:
+ if message.Address.Kind != protocol.Command {
+ return fmt.Errorf("delivery is not a command")
+ }
+
+ commands = []Command{{
+ DeviceID: message.Address.ID,
+ Capability: message.Address.Target,
+ Metadata: message.Metadata,
+ Payload: payload,
+ }}
+ case protocol.BatchPayload:
+ for _, member := range payload.Commands {
+ commands = append(commands, Command{
+ DeviceID: message.Address.ID,
+ Capability: member.Capability,
+ Metadata: member.Metadata,
+ Payload: member.Payload,
+ })
+ }
+ default:
+ return fmt.Errorf("delivery is not a command")
+ }
+
+ queue, exists := r.queues[message.Address.ID]
+ if !exists {
+ return fmt.Errorf("command device has no queue")
+ }
+
+ expiresAt := *commands[0].Metadata.ExpiresAt
+
+ for _, command := range commands {
+ if command.Expired(time.Now()) {
+ return fmt.Errorf("command expired")
+ }
+
+ if command.Metadata.ExpiresAt.Before(expiresAt) {
+ expiresAt = *command.Metadata.ExpiresAt
+ }
+ }
+
+ ctx, cancel := context.WithDeadline(ctx, expiresAt)
+ defer cancel()
+
+ err = ctx.Err()
+ if err != nil {
+ return err
+ }
+
+ select {
+ case queue <- commands:
+ return nil
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
--- /dev/null
+package driver
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestRouterKeepsSinglesAndBatchesInOneDeviceOrder(t *testing.T) {
+ router, err := NewRouter([]string{testDeviceID})
+ require.NoError(t, err)
+
+ queue, exists := router.Queue(testDeviceID)
+ require.True(t, exists)
+
+ first := routerCommand(t, testDeviceID, "relay", time.Minute)
+ second := routerCommand(t, testDeviceID, "output", time.Minute)
+ third := routerCommand(t, testDeviceID, "pulse", time.Minute)
+ last := routerCommand(t, testDeviceID, "batch", time.Minute)
+ batch, err := protocol.BuildCommandBatch(testDeviceID, second.Metadata, protocol.BatchPayload{
+ Commands: []protocol.BatchCommand{
+ {
+ Metadata: second.Metadata,
+ Capability: second.Capability,
+ Payload: second.Payload,
+ },
+ {
+ Metadata: third.Metadata,
+ Capability: third.Capability,
+ Payload: third.Payload,
+ },
+ },
+ })
+ require.NoError(t, err)
+
+ for _, publication := range []protocol.Publication{commandPublication(t, first), batch, commandPublication(t, last)} {
+ require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, publication.Retain))
+ }
+
+ require.Len(t, queue, 3)
+ require.Equal(t, []Command{first}, <-queue)
+ require.Equal(t, []Command{second, third}, <-queue)
+ require.Equal(t, []Command{last}, <-queue)
+}
+
+func TestRouterBuildsOnlyExplicitDeviceSubscriptions(t *testing.T) {
+ router, err := NewRouter([]string{testOtherDeviceID, testDeviceID})
+ require.NoError(t, err)
+
+ first, err := protocol.CommandSubscription(testDeviceID)
+ require.NoError(t, err)
+
+ second, err := protocol.CommandSubscription(testOtherDeviceID)
+ require.NoError(t, err)
+
+ subscriptions := router.Subscriptions()
+ require.Equal(t, []protocol.Subscription{first, second}, subscriptions)
+
+ subscriptions[0].Filter = "changed"
+
+ require.Equal(t, first, router.Subscriptions()[0])
+
+ queue, exists := router.Queue("unknown")
+ require.False(t, exists)
+ require.Nil(t, queue)
+
+ _, err = NewRouter([]string{"invalid"})
+ require.Error(t, err)
+
+ _, err = NewRouter([]string{testDeviceID, testDeviceID})
+ require.Error(t, err)
+
+ readOnly, err := NewRouter(nil)
+ require.NoError(t, err)
+ require.Empty(t, readOnly.Subscriptions())
+}
+
+func TestRouterRejectsInvalidUnownedAndExpiredDeliveries(t *testing.T) {
+ router, err := NewRouter([]string{testDeviceID})
+ require.NoError(t, err)
+
+ queue, exists := router.Queue(testDeviceID)
+ require.True(t, exists)
+
+ valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Minute))
+ unowned := commandPublication(t, routerCommand(t, testOtherDeviceID, "relay", time.Minute))
+ expired := commandPublication(t, routerCommand(t, testDeviceID, "relay", -time.Second))
+ value, err := protocol.BuildReported(testDeviceID, "relay", protocolMetadata(t), protocol.ValuePayload{Value: json.RawMessage("true")})
+ require.NoError(t, err)
+
+ availability, err := protocol.BuildAvailability(testDeviceID, protocolMetadata(t), protocol.AvailabilityPayload{Status: "online"})
+ require.NoError(t, err)
+
+ testCases := []struct {
+ name string
+ publication protocol.Publication
+ }{
+ {
+ name: "retained command rejects",
+ publication: protocol.Publication{
+ Topic: valid.Topic,
+ Data: valid.Data,
+ Retain: true,
+ },
+ },
+ {
+ name: "malformed payload rejects",
+ publication: protocol.Publication{
+ Topic: valid.Topic,
+ Data: []byte(`{}`),
+ },
+ },
+ {
+ name: "unknown topic rejects",
+ publication: protocol.Publication{
+ Topic: "invalid",
+ Data: valid.Data,
+ },
+ },
+ {
+ name: "unowned device rejects",
+ publication: unowned,
+ },
+ {
+ name: "expired command rejects",
+ publication: expired,
+ },
+ {
+ name: "reported value is not command intake",
+ publication: value,
+ },
+ {
+ name: "availability is not command intake",
+ publication: availability,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ require.Error(t, router.Handle(t.Context(), tc.publication.Topic, tc.publication.Data, tc.publication.Retain))
+ require.Empty(t, queue)
+ })
+ }
+}
+
+func TestRouterRejectsWholeBatchWhenOneMemberExpires(t *testing.T) {
+ router, err := NewRouter([]string{testDeviceID})
+ require.NoError(t, err)
+
+ first := routerCommand(t, testDeviceID, "relay", time.Minute)
+ second := routerCommand(t, testDeviceID, "pulse", -time.Second)
+ batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{
+ Commands: []protocol.BatchCommand{
+ {
+ Metadata: first.Metadata,
+ Capability: first.Capability,
+ Payload: first.Payload,
+ },
+ {
+ Metadata: second.Metadata,
+ Capability: second.Capability,
+ Payload: second.Payload,
+ },
+ },
+ })
+ require.NoError(t, err)
+ require.Error(t, router.Handle(t.Context(), batch.Topic, batch.Data, false))
+
+ queue, exists := router.Queue(testDeviceID)
+ require.True(t, exists)
+ require.Empty(t, queue)
+}
+
+func TestRouterBackpressureHonorsCancellationAndEarliestExpiry(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ router, err := NewRouter([]string{testDeviceID})
+ require.NoError(t, err)
+
+ valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Hour))
+
+ for range commandQueueCapacity {
+ require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
+ }
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ finished := make(chan error, 1)
+ go func() {
+ finished <- router.Handle(ctx, valid.Topic, valid.Data, false)
+ }()
+
+ synctest.Wait()
+ require.Empty(t, finished)
+ cancel()
+ require.ErrorIs(t, <-finished, context.Canceled)
+
+ short := commandPublication(t, routerCommand(t, testDeviceID, "pulse", time.Second))
+ require.ErrorIs(t, router.Handle(t.Context(), short.Topic, short.Data, false), context.DeadlineExceeded)
+
+ first := routerCommand(t, testDeviceID, "relay", time.Minute)
+ second := routerCommand(t, testDeviceID, "pulse", time.Second)
+ batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{Commands: []protocol.BatchCommand{
+ {
+ Metadata: first.Metadata,
+ Capability: first.Capability,
+ Payload: first.Payload,
+ },
+ {
+ Metadata: second.Metadata,
+ Capability: second.Capability,
+ Payload: second.Payload,
+ },
+ }})
+ require.NoError(t, err)
+
+ start := time.Now()
+
+ require.ErrorIs(t, router.Handle(t.Context(), batch.Topic, batch.Data, false), context.DeadlineExceeded)
+ require.Equal(t, time.Second, time.Since(start))
+
+ queue, exists := router.Queue(testDeviceID)
+ require.True(t, exists)
+ require.Len(t, queue, commandQueueCapacity)
+ require.ErrorIs(t, router.Handle(ctx, valid.Topic, valid.Data, false), context.Canceled)
+ <-queue
+ require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
+ })
+}
+
+func routerCommand(t *testing.T, deviceID string, capability string, lifetime time.Duration) Command {
+ t.Helper()
+
+ metadata := protocolMetadata(t)
+ expiry := time.Now().UTC().Add(lifetime)
+ metadata.ExpiresAt = &expiry
+
+ return Command{
+ DeviceID: deviceID,
+ Capability: capability,
+ Metadata: metadata,
+ Payload: protocol.ValuePayload{Value: json.RawMessage("true")},
+ }
+}
+
+func commandPublication(t *testing.T, command Command) protocol.Publication {
+ t.Helper()
+
+ publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+ require.NoError(t, err)
+
+ return publication
+}
+
+func protocolMetadata(t *testing.T) protocol.Metadata {
+ t.Helper()
+
+ now := time.Now().UTC()
+ id, err := protocol.NewID(now)
+ require.NoError(t, err)
+
+ return protocol.Metadata{
+ MessageID: id,
+ SentAt: &now,
+ }
+}
return errors.Join(failures...)
}
-type driverResult struct {
- index int
- err error
-}
-
func stopOnly(err error, cancellation error) bool {
joined, ok := err.(interface{ Unwrap() []error })
if ok {
projects / onehouse / commitdiff