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