git.giacomolli.uk - onehouse/blob - protocol/address.go

   1 package protocol
   2
   3 import (
   4         "fmt"
   5         "strings"
   6 )
   7
   8 const (
   9         // Root is the versioned MQTT namespace.
  10         Root = "onehouse/v1"
  11
  12         // Descriptor identifies retained device metadata.
  13         Descriptor Kind = "descriptor"
  14
  15         // Availability identifies retained device reachability.
  16         Availability Kind = "availability"
  17
  18         // Reported identifies a retained point value.
  19         Reported Kind = "reported"
  20
  21         // Telemetry identifies a non-retained measurement.
  22         Telemetry Kind = "telemetry"
  23
  24         // Event identifies a non-retained occurrence.
  25         Event Kind = "events"
  26
  27         // Command identifies a non-retained point request.
  28         Command Kind = "commands"
  29
  30         // CommandBatch identifies an ordered group of point requests.
  31         CommandBatch Kind = "batch"
  32
  33         // Result identifies a command outcome.
  34         Result Kind = "results"
  35
  36         // HubStatus identifies retained Zigbee hub diagnostics.
  37         HubStatus Kind = "status"
  38 )
  39
  40 // Kind identifies a wire operation, independent of its topic text.
  41 type Kind string
  42
  43 // Address identifies a device operation or a Zigbee hub status. Target is a
  44 // capability, event, or command ID. Hub status and device metadata omit Target.
  45 type Address struct {
  46         Kind   Kind
  47         ID     string
  48         Target string
  49 }
  50
  51 // topic validates the address and returns its exact publication topic.
  52 func (a Address) topic() (string, error) {
  53         if !ValidID(a.ID) {
  54                 return "", fmt.Errorf("address ID must be a canonical ULID")
  55         }
  56
  57         switch a.Kind {
  58         case Descriptor, Availability:
  59                 if a.Target != "" {
  60                         return "", fmt.Errorf("metadata address must not have a target")
  61                 }
  62
  63                 return Root + "/devices/" + a.ID + "/" + string(a.Kind), nil
  64         case Reported, Telemetry, Event, Command:
  65                 if !ValidIdentifier(a.Target) {
  66                         return "", fmt.Errorf("point address must have a valid target")
  67                 }
  68         case Result:
  69                 if !ValidID(a.Target) {
  70                         return "", fmt.Errorf("result target must be a canonical ULID")
  71                 }
  72         case CommandBatch:
  73                 if a.Target != "" {
  74                         return "", fmt.Errorf("batch address must not have a target")
  75                 }
  76
  77                 return Root + "/devices/" + a.ID + "/commands/$batch", nil
  78         case HubStatus:
  79                 if a.Target != "" {
  80                         return "", fmt.Errorf("hub address must not have a target")
  81                 }
  82
  83                 return Root + "/zigbee/hubs/" + a.ID + "/status", nil
  84         default:
  85                 return "", fmt.Errorf("unknown address kind %q", a.Kind)
  86         }
  87
  88         return Root + "/devices/" + a.ID + "/" + string(a.Kind) + "/" + a.Target, nil
  89 }
  90
  91 // parseTopic rejects malformed topics, wildcards, and unsupported operations.
  92 func parseTopic(topic string) (Address, error) {
  93         parts := strings.Split(topic, "/")
  94         if len(parts) < 5 || parts[0]+"/"+parts[1] != Root {
  95                 return Address{}, fmt.Errorf("topic must use the versioned root")
  96         }
  97
  98         var address Address
  99
 100         switch {
 101         case len(parts) == 6 && parts[2] == "zigbee" && parts[3] == "hubs" && parts[5] == "status":
 102                 address = Address{
 103                         Kind: HubStatus,
 104                         ID:   parts[4],
 105                 }
 106         case (len(parts) == 5 || len(parts) == 6) && parts[2] == "devices":
 107                 address = Address{
 108                         Kind: Kind(parts[4]),
 109                         ID:   parts[3],
 110                 }
 111                 if len(parts) == 6 {
 112                         address.Target = parts[5]
 113                         if address.Kind == Command && address.Target == "$batch" {
 114                                 address.Kind = CommandBatch
 115                                 address.Target = ""
 116                         }
 117                 }
 118         default:
 119                 return Address{}, fmt.Errorf("topic layout is invalid")
 120         }
 121
 122         canonical, err := address.topic()
 123         if err != nil {
 124                 return Address{}, fmt.Errorf("validate topic address: %w", err)
 125         }
 126
 127         if topic != canonical {
 128                 return Address{}, fmt.Errorf("topic is not canonical")
 129         }
 130
 131         return address, nil
 132 }
 133
 134 // Subscription contains a topic filter and the requested delivery level.
 135 type Subscription struct {
 136         Filter string
 137         QoS    byte
 138 }
 139
 140 // CoreSubscriptions returns device-to-core filters. Hub diagnostics are not
 141 // house-state input. Telemetry uses QoS 0. Other subscriptions use QoS 1.
 142 func CoreSubscriptions() []Subscription {
 143         return []Subscription{
 144                 {
 145                         Filter: Root + "/devices/+/descriptor",
 146                         QoS:    1,
 147                 },
 148                 {
 149                         Filter: Root + "/devices/+/availability",
 150                         QoS:    1,
 151                 },
 152                 {
 153                         Filter: Root + "/devices/+/reported/+",
 154                         QoS:    1,
 155                 },
 156                 {
 157                         Filter: Root + "/devices/+/telemetry/+",
 158                         QoS:    0,
 159                 },
 160                 {
 161                         Filter: Root + "/devices/+/events/+",
 162                         QoS:    1,
 163                 },
 164                 {
 165                         Filter: Root + "/devices/+/results/+",
 166                         QoS:    1,
 167                 },
 168         }
 169 }
 170
 171 // CommandSubscription returns the device command filter, including batches.
 172 func CommandSubscription(deviceID string) (Subscription, error) {
 173         if !ValidID(deviceID) {
 174                 return Subscription{}, fmt.Errorf("device ID must be a canonical ULID")
 175         }
 176
 177         return Subscription{
 178                 Filter: Root + "/devices/" + deviceID + "/commands/+",
 179                 QoS:    1,
 180         }, nil
 181 }
 182
 183 // HubStatusSubscription returns the filter for all Zigbee hub diagnostics.
 184 func HubStatusSubscription() Subscription {
 185         return Subscription{
 186                 Filter: Root + "/zigbee/hubs/+/status",
 187                 QoS:    1,
 188         }
 189 }