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

   1 package m31
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "net"
   8         "strconv"
   9         "time"
  10
  11         "onehouse/driver"
  12         "onehouse/protocol"
  13 )
  14
  15 // Driver polls one assembly. Run has one owner and closes the supplied Client
  16 // on each attempt. Recoverable device failures publish offline and retry with
  17 // backoff. Publication or cleanup failures return to the framework runner. The
  18 // shared publisher must run alongside this driver. Refresh is safe during Run.
  19 // No operation starts a physical connection before Run.
  20 type Driver struct {
  21         deviceID     string
  22         address      string
  23         unit         byte
  24         pollInterval time.Duration
  25         topology     topology
  26         publisher    *driver.Publisher
  27         client       Client
  28 }
  29
  30 // New validates settings. The caller supplies valid publisher and client
  31 // dependencies. The driver owns the client until Run returns.
  32 func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver, error) {
  33         if !protocol.ValidID(settings.DeviceID) {
  34                 return nil, fmt.Errorf("device ID must be a canonical ULID")
  35         }
  36
  37         host, port, err := net.SplitHostPort(settings.Address)
  38         if err != nil || host == "" {
  39                 return nil, fmt.Errorf("address must contain a host and TCP port")
  40         }
  41
  42         number, err := strconv.Atoi(port)
  43         if err != nil || number < 1 || number > 65535 {
  44                 return nil, fmt.Errorf("TCP port is outside its range")
  45         }
  46
  47         if settings.Unit == 0 || settings.Unit > 247 {
  48                 return nil, fmt.Errorf("unit must be from one through 247")
  49         }
  50
  51         if settings.PollInterval < 10*time.Millisecond || settings.PollInterval > time.Minute {
  52                 return nil, fmt.Errorf("poll interval must be from ten milliseconds through one minute")
  53         }
  54
  55         points, err := newTopology(settings.Modules)
  56         if err != nil {
  57                 return nil, fmt.Errorf("validate topology: %w", err)
  58         }
  59
  60         return &Driver{
  61                 deviceID:     settings.DeviceID,
  62                 address:      settings.Address,
  63                 unit:         settings.Unit,
  64                 pollInterval: settings.PollInterval,
  65                 topology:     points,
  66                 publisher:    publisher,
  67                 client:       client,
  68         }, nil
  69 }
  70
  71 // Run configures and verifies output modes before online or polling. A
  72 // successful first read establishes a baseline without input events. Device
  73 // failures retry after one second, doubling to thirty seconds until a complete
  74 // baseline succeeds.
  75 func (d *Driver) Run(ctx context.Context) error {
  76         err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
  77         if err != nil {
  78                 return fmt.Errorf("publish descriptor: %w", err)
  79         }
  80
  81         delay := time.Second
  82
  83         for {
  84                 healthy, retry, err := d.attempt(ctx)
  85                 if !retry || ctx.Err() != nil {
  86                         return fmt.Errorf("run assembly: %w", err)
  87                 }
  88
  89                 err = d.publisher.Availability(ctx, d.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
  90                 if err != nil {
  91                         return fmt.Errorf("publish offline: %w", err)
  92                 }
  93
  94                 if healthy {
  95                         delay = time.Second
  96                 }
  97
  98                 err = d.wait(ctx, delay)
  99                 if err != nil {
 100                         return fmt.Errorf("wait for reconnect: %w", err)
 101                 }
 102
 103                 delay = min(delay*2, 30*time.Second)
 104         }
 105 }
 106
 107 // Refresh requests retained snapshots through the shared publisher. It does not
 108 // repeat input events, physical commands, or pulse operations.
 109 func (d *Driver) Refresh() {
 110         d.publisher.Refresh()
 111 }
 112
 113 func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err error) {
 114         defer func() {
 115                 var failure *deviceError
 116
 117                 retry = errors.As(err, &failure)
 118
 119                 closeErr := d.client.Close()
 120                 if closeErr != nil {
 121                         retry = false
 122                         err = fmt.Errorf("close assembly: %w", errors.Join(err, closeErr))
 123                 }
 124         }()
 125
 126         err = d.client.Connect(ctx, d.address, d.unit)
 127         if err != nil {
 128                 return false, false, &deviceError{err: fmt.Errorf("connect assembly: %w", err)}
 129         }
 130
 131         connection := &connection{owner: d}
 132         err = connection.start(ctx)
 133         if err != nil {
 134                 return false, false, fmt.Errorf("initialize assembly: %w", err)
 135         }
 136
 137         healthy = true
 138         ticker := time.NewTicker(d.pollInterval)
 139         defer ticker.Stop()
 140
 141         for {
 142                 select {
 143                 case <-ctx.Done():
 144                         return healthy, false, ctx.Err()
 145                 case <-ticker.C:
 146                         err = connection.poll(ctx)
 147                         if err != nil {
 148                                 return healthy, false, fmt.Errorf("poll assembly: %w", err)
 149                         }
 150                 }
 151         }
 152 }
 153
 154 func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
 155         timer := time.NewTimer(delay)
 156         defer timer.Stop()
 157
 158         select {
 159         case <-ctx.Done():
 160                 return ctx.Err()
 161         case <-timer.C:
 162                 return nil
 163         }
 164 }