git.giacomolli.uk - onehouse/blob - driver/router.go

   1 package driver
   2
   3 import (
   4         "context"
   5         "fmt"
   6         "slices"
   7         "time"
   8
   9         "onehouse/protocol"
  10 )
  11
  12 const commandQueueCapacity = 64
  13
  14 // Router owns one bounded FIFO for each supplied command-capable device. A
  15 // queue entry is one single command or one ordered batch. Handle is safe for
  16 // concurrent calls. Each queue has one consumer that serializes device work.
  17 // The caller stops producers and consumers through their contexts. Queues do
  18 // not close and belong to the router for its lifetime.
  19 type Router struct {
  20         queues        map[string]chan []Command
  21         subscriptions []protocol.Subscription
  22 }
  23
  24 // NewRouter validates unique device IDs and makes 64-entry queues. Read-only
  25 // devices do not belong in deviceIDs. The constructor performs no I/O.
  26 func NewRouter(deviceIDs []string) (*Router, error) {
  27         ids := slices.Clone(deviceIDs)
  28         slices.Sort(ids)
  29
  30         router := &Router{queues: make(map[string]chan []Command, len(deviceIDs))}
  31
  32         for _, deviceID := range ids {
  33                 _, exists := router.queues[deviceID]
  34                 if exists {
  35                         return nil, fmt.Errorf("duplicate command device ID")
  36                 }
  37
  38                 subscription, err := protocol.CommandSubscription(deviceID)
  39                 if err != nil {
  40                         return nil, fmt.Errorf("build command subscription: %w", err)
  41                 }
  42
  43                 router.queues[deviceID] = make(chan []Command, commandQueueCapacity)
  44                 router.subscriptions = append(router.subscriptions, subscription)
  45         }
  46
  47         return router, nil
  48 }
  49
  50 // Subscriptions returns an independent, sorted list including batch intake.
  51 func (r *Router) Subscriptions() []protocol.Subscription {
  52         return slices.Clone(r.subscriptions)
  53 }
  54
  55 // Queue returns the receive-only command queue for a configured device.
  56 func (r *Router) Queue(deviceID string) (<-chan []Command, bool) {
  57         queue, exists := r.queues[deviceID]
  58
  59         return queue, exists
  60 }
  61
  62 // Handle parses one delivery once and queues all its commands together. It
  63 // rejects malformed, retained, expired, non-command, and unowned deliveries. A
  64 // full queue waits until space, caller cancellation, or the earliest member
  65 // expiry. A driver must check expiry again when it consumes the queue entry.
  66 func (r *Router) Handle(ctx context.Context, topic string, data []byte, retained bool) error {
  67         err := ctx.Err()
  68         if err != nil {
  69                 return err
  70         }
  71
  72         message, err := protocol.Parse(topic, data, retained)
  73         if err != nil {
  74                 return fmt.Errorf("parse command delivery: %w", err)
  75         }
  76
  77         var commands []Command
  78
  79         switch payload := message.Payload.(type) {
  80         case protocol.ValuePayload:
  81                 if message.Address.Kind != protocol.Command {
  82                         return fmt.Errorf("delivery is not a command")
  83                 }
  84
  85                 commands = []Command{{
  86                         DeviceID:   message.Address.ID,
  87                         Capability: message.Address.Target,
  88                         Metadata:   message.Metadata,
  89                         Payload:    payload,
  90                 }}
  91         case protocol.BatchPayload:
  92                 for _, member := range payload.Commands {
  93                         commands = append(commands, Command{
  94                                 DeviceID:   message.Address.ID,
  95                                 Capability: member.Capability,
  96                                 Metadata:   member.Metadata,
  97                                 Payload:    member.Payload,
  98                         })
  99                 }
 100         default:
 101                 return fmt.Errorf("delivery is not a command")
 102         }
 103
 104         queue, exists := r.queues[message.Address.ID]
 105         if !exists {
 106                 return fmt.Errorf("command device has no queue")
 107         }
 108
 109         expiresAt := *commands[0].Metadata.ExpiresAt
 110
 111         for _, command := range commands {
 112                 if command.Expired(time.Now()) {
 113                         return fmt.Errorf("command expired")
 114                 }
 115
 116                 if command.Metadata.ExpiresAt.Before(expiresAt) {
 117                         expiresAt = *command.Metadata.ExpiresAt
 118                 }
 119         }
 120
 121         ctx, cancel := context.WithDeadline(ctx, expiresAt)
 122         defer cancel()
 123
 124         err = ctx.Err()
 125         if err != nil {
 126                 return err
 127         }
 128
 129         select {
 130         case queue <- commands:
 131                 return nil
 132         case <-ctx.Done():
 133                 return ctx.Err()
 134         }
 135 }