equipment-operator

1package main23// The wire types are hand-written, the way liken and the sibling4// operators write theirs. The Kubernetes API is HTTPS that serves5// JSON, and importing client-go for a dozen structs brings informers,6// work queues, and a release cadence this program does not use. Each7// type carries only the fields this operator reads or writes; the8// API server fills in the rest.910// The group this operator serves.11const equipmentAPIVersion = "equipment.liken.sh/v1alpha1"1213// ObjectMeta carries what this operator reads or writes: name for the14// URL, and resourceVersion for the conditional write.15type ObjectMeta struct {16	Name            string `json:"name,omitempty"`17	ResourceVersion string `json:"resourceVersion,omitempty"`18	Generation      int64  `json:"generation,omitempty"`19}2021// A list's own resourceVersion is the revision of the whole22// collection, which is what a watch resumes from.23type ListMeta struct {24	ResourceVersion string `json:"resourceVersion,omitempty"`25}2627// A Receiver is one piece of A/V equipment on the far end of a28// machine's cable. It is cluster-scoped, because the machines that feed29// it are the cluster's.30type Receiver struct {31	APIVersion string         `json:"apiVersion,omitempty"`32	Kind       string         `json:"kind,omitempty"`33	Metadata   ObjectMeta     `json:"metadata"`34	Spec       ReceiverSpec   `json:"spec"`35	Status     ReceiverStatus `json:"status,omitempty"`36}3738type ReceiverList struct {39	Metadata ListMeta   `json:"metadata"`40	Items    []Receiver `json:"items"`41}4243// The spec names one protocol block, the inputs liken machines feed,44// and the session that holds the receiver now.45type ReceiverSpec struct {46	Denon   *DenonProtocol   `json:"denon,omitempty"`47	Volume  *ReceiverVolume  `json:"volume,omitempty"`48	Inputs  []ReceiverInput  `json:"inputs,omitempty"`49	Session *ReceiverSession `json:"session,omitempty"`50}5152// How loud the room may get and how far one press moves it, both in the53// receiver's own scale. A Denon requires max. An absent step is one54// whole unit of that scale.55type ReceiverVolume struct {56	Max  float64 `json:"max,omitempty"`57	Step float64 `json:"step,omitempty"`58}5960// The Denon and Marantz control protocol, and the address it answers61// on.62type DenonProtocol struct {63	Address string `json:"address"`64}6566// One input of the receiver, and the machine and monitor id that feed67// it.68type ReceiverInput struct {69	Name    string `json:"name"`70	Machine string `json:"machine"`71	Monitor string `json:"monitor"`72}7374// What a session names: the Player, the input it plays through, the75// topic it takes the level from, and the two flags the media operator76// flips on it. Active says a Play stands. Awake says the room's screen77// is awake. The media operator holds a session whenever the Player has78// a screen, the idle screen included, so a session with both flags off79// owns the level and sends the equipment nothing.80type ReceiverSession struct {81	Player      string `json:"player"`82	Input       string `json:"input"`83	VolumeTopic string `json:"volumeTopic"`84	Active      bool   `json:"active,omitempty"`85	Awake       bool   `json:"awake,omitempty"`86}8788// withoutFlags is the session apart from Active and Awake, which is89// what tells one session from another. A flip of either flag reaches90// the session that already stands. Anything else replaces it.91func (s ReceiverSession) withoutFlags() ReceiverSession {92	s.Active = false93	s.Awake = false94	return s95}9697// What the receiver last said, in its own units, and the Reachable98// condition.99type ReceiverStatus struct {100	Power      string      `json:"power,omitempty"`101	Input      string      `json:"input,omitempty"`102	Volume     string      `json:"volume,omitempty"`103	VolumeMax  string      `json:"volumeMax,omitempty"`104	Mute       bool        `json:"mute,omitempty"`105	SoundMode  string      `json:"soundMode,omitempty"`106	Service    string      `json:"service,omitempty"`107	Conditions []Condition `json:"conditions,omitempty"`108}109110// ConditionStatus is the three-valued verdict a condition carries.111type ConditionStatus string112113const (114	ConditionTrue    ConditionStatus = "True"115	ConditionFalse   ConditionStatus = "False"116	ConditionUnknown ConditionStatus = "Unknown"117)118119// Condition mirrors metav1.Condition, the shape Kubernetes uses120// everywhere, and liken's own. Anyone who reads kubectl describe121// output on a Pod already knows how to read one of these.122//123// ObservedGeneration records which metadata.generation the condition124// judged. Generation counts spec edits, so a reader can tell "Reachable,125// for the spec as it stands" from "Reachable, but for a spec two edits126// ago".127type Condition struct {128	Type               string          `json:"type"`129	Status             ConditionStatus `json:"status"`130	ObservedGeneration int64           `json:"observedGeneration,omitempty"`131	Reason             string          `json:"reason,omitempty"`132	Message            string          `json:"message,omitempty"`133	LastTransitionTime string          `json:"lastTransitionTime"`134}

1package main23// This is a Kubernetes client written straight against the HTTP API,4// following liken's own (kubernetes/apiclient.go) and the media5// operator's, for the same reason: the API is HTTPS that serves6// JSON, and client-go would bring informers, work queues, and7// generated types this program does not use.8//9// Every pod already holds what it needs to reach the API server.10// Kubernetes injects two environment variables that name the11// server's in-cluster address, and the kubelet mounts a CA12// certificate and a ServiceAccount token at a known path. Those five13// values are the whole of an in-cluster config.1415import (16	"bytes"17	"context"18	"crypto/tls"19	"crypto/x509"20	"encoding/json"21	"errors"22	"fmt"23	"io"24	"net"25	"net/http"26	"os"27	"time"28)2930// serviceAccountDir is a variable so a test points it at a directory31// it controls.32var serviceAccountDir = "/var/run/secrets/kubernetes.io/serviceaccount"3334// These two answers are values, not failures. An absent object is35// the normal state the caller answers by creating it, and a conflict36// is the normal state under optimistic concurrency that the caller37// answers by reading again.38var (39	ErrNotFound = errors.New("not found")40	ErrConflict = errors.New("conflict: something else wrote this object first")41)4243type Client struct {44	base        string45	http        *http.Client46	credentials string47}4849// NewClient builds a client from its three parts. InClusterClient50// reads them from the pod's environment; a test hands in an51// httptest server's base and no credentials.52func NewClient(base string, httpClient *http.Client, credentials string) *Client {53	return &Client{base: base, http: httpClient, credentials: credentials}54}5556func InClusterClient() (*Client, error) {57	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")58	if host == "" || port == "" {59		return nil, fmt.Errorf("not running in a cluster: KUBERNETES_SERVICE_HOST unset")60	}6162	// The client trusts the cluster's own CA and not the system63	// store, so it accepts this API server and no other server that64	// answers on the address.65	caPEM, err := os.ReadFile(serviceAccountDir + "/ca.crt")66	if err != nil {67		return nil, fmt.Errorf("reading service account CA: %w", err)68	}69	roots := x509.NewCertPool()70	if !roots.AppendCertsFromPEM(caPEM) {71		return nil, fmt.Errorf("service account CA contains no certificates")72	}7374	return NewClient("https://"+host+":"+port, &http.Client{75		Transport: &http.Transport{76			TLSClientConfig: &tls.Config{RootCAs: roots},77			// Each timeout bounds the same failure: a server that78			// stops answering without sending anything. There is no79			// overall client timeout, because the watch is a request80			// whose response never ends, and a whole-request81			// deadline would cut the stream on schedule.82			DialContext: (&net.Dialer{83				Timeout:   5 * time.Second,84				KeepAlive: 10 * time.Second,85			}).DialContext,86			ResponseHeaderTimeout: 10 * time.Second,87			IdleConnTimeout:       30 * time.Second,88		},89	}, serviceAccountDir), nil90}9192// The two content types this client sends. A body is JSON,93// except an apply, which the API server reads as a partial object94// under the caller's field manager. The apply media type is named for95// YAML and accepts JSON, because YAML is its superset.96const (97	jsonContentType  = "application/json"98	applyContentType = "application/apply-patch+yaml"99)100101// fieldManager names this operator's writes to the API server, so102// server-side apply gives it the fields it applies and leaves every103// other writer's fields alone.104const fieldManager = "equipment-operator"105106// Do sends one request and hands back the open response, which is107// what the watch needs and what RequestJSON is built on. The context108// governs the whole exchange, the response body included, so109// cancelling it is what ends a read of a stream that never ends.110func (c *Client) Do(ctx context.Context, method, path string, body []byte) (*http.Response, error) {111	return c.send(ctx, method, path, jsonContentType, body)112}113114func (c *Client) send(ctx context.Context, method, path, contentType string, body []byte) (*http.Response, error) {115	var reader io.Reader116	if body != nil {117		reader = bytes.NewReader(body)118	}119	req, err := http.NewRequestWithContext(ctx, method, c.base+path, reader)120	if err != nil {121		return nil, err122	}123	// The token is read from disk on every request. The mounted124	// token is short-lived and the kubelet refreshes the file as125	// each one nears expiry, so a client that held one in memory126	// would start getting 401s.127	if c.credentials != "" {128		token, err := os.ReadFile(c.credentials + "/token")129		if err != nil {130			return nil, fmt.Errorf("reading service account token: %w", err)131		}132		req.Header.Set("Authorization", "Bearer "+string(token))133	}134	req.Header.Set("Accept", "application/json")135	if body != nil {136		req.Header.Set("Content-Type", contentType)137	}138	return c.http.Do(req)139}140141// RequestJSON sends one request and decodes the answer, turning142// every non-2xx status into an error that carries the server's own143// message.144func (c *Client) RequestJSON(method, path string, body []byte, out any) error {145	return c.requestJSON(method, path, jsonContentType, body, out)146}147148func (c *Client) requestJSON(method, path, contentType string, body []byte, out any) error {149	resp, err := c.send(context.Background(), method, path, contentType, body)150	if err != nil {151		return err152	}153	defer drain(resp.Body)154155	if resp.StatusCode == http.StatusNotFound {156		return ErrNotFound157	}158	if resp.StatusCode == http.StatusConflict {159		return ErrConflict160	}161	if resp.StatusCode < 200 || resp.StatusCode > 299 {162		message, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))163		return fmt.Errorf("%s %s: %s: %s", method, path, resp.Status, message)164	}165	return json.NewDecoder(resp.Body).Decode(out)166}167168// drain reads whatever the caller left in the body, then closes it.169// Go returns a connection to its pool only when the body reaches170// EOF, so an early close costs a fresh connection and TLS handshake,171// and reaches the server as a hang-up on a request it answered.172const maxDrain = 4 << 20173174func drain(body io.ReadCloser) {175	_, _ = io.Copy(io.Discard, io.LimitReader(body, maxDrain))176	_ = body.Close()177}178179// A Receiver is cluster-scoped, so the collection path carries no180// namespace, the way a StorageClass path carries none.181const receiversPath = "/apis/" + equipmentAPIVersion + "/receivers"182183func receiverPath(name string) string {184	return receiversPath + "/" + name185}186187// ListReceivers answers a whole pass with one request, and the list's188// resourceVersion is where a watch resumes from.189func ListReceivers(c *Client) (*ReceiverList, error) {190	list := &ReceiverList{}191	if err := c.RequestJSON(http.MethodGet, receiversPath, nil, list); err != nil {192		return nil, err193	}194	return list, nil195}196197func GetReceiver(c *Client, name string) (*Receiver, error) {198	receiver := &Receiver{}199	if err := c.RequestJSON(http.MethodGet, receiverPath(name), nil, receiver); err != nil {200		return nil, err201	}202	return receiver, nil203}204205// WatchReceivers opens the stream the loop wakes on. A watch is an206// ordinary GET whose response never ends: the API server holds the207// connection open and writes one JSON event per change. The caller208// owns the body, and resumes from the resourceVersion a list gave it,209// so no change is missed between reconnects. Bookmarks cost one line210// each and keep the resume point current while nothing changes.211func WatchReceivers(ctx context.Context, c *Client, resourceVersion string) (*http.Response, error) {212	path := receiversPath + "?watch=true&allowWatchBookmarks=true&resourceVersion=" + resourceVersion213	return c.Do(ctx, http.MethodGet, path, nil)214}215216// receiverStatusApply is the partial object an apply sends: the217// identity the API server matches on, and the status this operator218// owns. It carries no spec, so an apply can never state a field a219// person declared.220type receiverStatusApply struct {221	APIVersion string         `json:"apiVersion"`222	Kind       string         `json:"kind"`223	Metadata   ObjectMeta     `json:"metadata"`224	Status     ReceiverStatus `json:"status"`225}226227// ApplyReceiverStatus writes the status subresource under this228// operator's own field manager. Server-side apply keeps the write to229// the fields the body states, and removes the fields this manager230// owned and no longer states, so a reading that stops arriving leaves231// no stale value behind. force settles a conflict in this manager's232// favour, because nothing else writes a Receiver's status.233func ApplyReceiverStatus(c *Client, name string, status ReceiverStatus) (*Receiver, error) {234	body, err := json.Marshal(&receiverStatusApply{235		APIVersion: equipmentAPIVersion,236		Kind:       "Receiver",237		Metadata:   ObjectMeta{Name: name},238		Status:     status,239	})240	if err != nil {241		return nil, err242	}243	path := receiverPath(name) + "/status?fieldManager=" + fieldManager + "&force=true"244	written := &Receiver{}245	if err := c.requestJSON(http.MethodPatch, path, applyContentType, body, written); err != nil {246		return nil, err247	}248	return written, nil249}

1package main23// The bus client is one live TCP connection to the broker, with a4// reader, a single writer, and a keepalive timer, and it reconnects5// with backoff whenever any of the three fails.6//7// One session holds one Bus, so no caller writes the socket directly.89import (10	"bufio"11	"context"12	"net"13	"sort"14	"sync"15	"time"16)1718// The keepalive the client asks the broker for, and the queue the19// writer drains. The client sends a PINGREQ once the connection has20// been idle longer than half the keepalive, so the broker never21// reaches the keepalive without hearing from the client. The queue is22// bounded, and a publish that would overflow it is dropped, which is23// correct at QoS 0.24const (25	busKeepalive  = 3026	busQueueDepth = 6427)2829// The reconnect backoff bounds. The client waits busMinBackoff after30// the first failure and doubles the wait up to busMaxBackoff, so a31// broker that is down does not become a tight reconnect loop. Both are32// variables so a test drives a reconnect in milliseconds.33var (34	busMinBackoff = time.Second35	busMaxBackoff = 30 * time.Second36)3738// busHandler receives one inbound message's topic and payload. The Bus39// calls it on the reader goroutine, so a handler that blocks holds up40// every later message on the connection.41type busHandler func(topic string, payload []byte)4243// busWill is the MQTT Last Will the client names at connect time. The44// broker publishes it on any disconnect the client does not make45// cleanly.46//47// busWill is the message the broker publishes when this connection dies48// without a clean disconnect. It is how a killed operator's owner mark49// is cleared.50type busWill struct {51	Topic    string52	Payload  []byte53	Retained bool54}5556// Bus holds the connection's parts and the state that outlives one57// connection: the remembered subscriptions and the packet identifier58// counter. out is the current connection's write queue, or nil while59// disconnected, and the mutex guards both it and the fields around it.60type Bus struct {61	clientID  string62	will      *busWill63	onConnect func(*Bus)64	handler   busHandler65	dial      func(context.Context) (net.Conn, error)6667	mutex    sync.Mutex68	filters  map[string]struct{}69	out      chan []byte70	packetID uint1671}7273// newBus builds a client that dials the address over TCP. The address,74// the client identifier, the will, the connect callback, and the75// inbound handler are fixed for the client's life; the connection they76// drive is not.77func newBus(address, clientID string, will *busWill, onConnect func(*Bus), handler busHandler) *Bus {78	bus := &Bus{79		clientID:  clientID,80		will:      will,81		onConnect: onConnect,82		handler:   handler,83		filters:   map[string]struct{}{},84	}85	bus.dial = func(ctx context.Context) (net.Conn, error) {86		dialer := &net.Dialer{}87		return dialer.DialContext(ctx, "tcp", address)88	}89	return bus90}9192// Run holds the connection open until ctx ends. It dials, connects,93// and serves one session, then waits a backoff and dials again. A94// session that reached a CONNACK resets the backoff to its floor, so a95// connection that drops after an hour reconnects at once, while a96// broker that never answers is retried ever more slowly.97func (b *Bus) Run(ctx context.Context) {98	backoff := busMinBackoff99	for ctx.Err() == nil {100		connected := b.runSession(ctx)101		if ctx.Err() != nil {102			return103		}104		if connected {105			backoff = busMinBackoff106		}107		select {108		case <-ctx.Done():109			return110		case <-time.After(backoff):111		}112		if !connected {113			backoff *= 2114			if backoff > busMaxBackoff {115				backoff = busMaxBackoff116			}117		}118	}119}120121// runSession dials, completes the CONNECT handshake, and serves the122// connection until its reader or writer fails. It returns whether the123// handshake reached a CONNACK, which is what tells Run to reset the124// backoff.125func (b *Bus) runSession(parent context.Context) (connected bool) {126	conn, err := b.dial(parent)127	if err != nil {128		return false129	}130	defer conn.Close()131132	if _, err := conn.Write(encodeConnect(b.clientID, busKeepalive, b.will)); err != nil {133		return false134	}135	reader := bufio.NewReader(conn)136	first, body, err := readPacket(reader)137	if err != nil || first&0xF0 != mqttConnack {138		return false139	}140	if err := parseConnack(body); err != nil {141		return false142	}143144	ctx, cancel := context.WithCancel(parent)145	defer cancel()146	// The reader blocks in Read until the broker writes or the147	// connection closes. Closing the connection when the session ends148	// is what unblocks a reader waiting on a silent broker.149	defer context.AfterFunc(ctx, func() { conn.Close() })()150151	out := make(chan []byte, busQueueDepth)152	b.mutex.Lock()153	b.out = out154	b.mutex.Unlock()155	defer func() {156		b.mutex.Lock()157		b.out = nil158		b.mutex.Unlock()159	}()160161	var writing sync.WaitGroup162	writing.Add(1)163	go func() {164		defer writing.Done()165		// A write failure ends the session, so the reader stops too.166		defer cancel()167		b.writeLoop(ctx, conn, out)168	}()169170	// The remembered subscriptions go out first, so the broker is171	// delivering again before onConnect re-publishes any retained172	// state.173	b.resubscribe(out)174	if b.onConnect != nil {175		b.onConnect(b)176	}177178	b.readLoop(reader)179	cancel()180	writing.Wait()181	return true182}183184// writeLoop is the one goroutine that writes the connection. It sends185// each queued frame and, when no frame has gone out for half the186// keepalive, sends a PINGREQ so the broker hears from the client187// before the keepalive elapses.188func (b *Bus) writeLoop(ctx context.Context, conn net.Conn, out <-chan []byte) {189	idle := time.Duration(busKeepalive) * time.Second / 2190	ticker := time.NewTicker(idle)191	defer ticker.Stop()192	last := time.Now()193	for {194		select {195		case <-ctx.Done():196			return197		case frame := <-out:198			if _, err := conn.Write(frame); err != nil {199				return200			}201			last = time.Now()202		case <-ticker.C:203			if time.Since(last) >= idle {204				if _, err := conn.Write(encodePingreq()); err != nil {205					return206				}207				last = time.Now()208			}209		}210	}211}212213// readLoop reads whole packets and delivers each inbound PUBLISH to the214// handler. A SUBACK and a PINGRESP are read and dropped: the client215// asks for QoS 0, so a SUBACK carries nothing to act on, and a216// PINGRESP only proves the broker is alive, which a successful read217// already shows. Any read error ends the loop and the session.218func (b *Bus) readLoop(reader *bufio.Reader) {219	for {220		first, body, err := readPacket(reader)221		if err != nil {222			return223		}224		if first&0xF0 == mqttPublish {225			topic, payload, ok := parsePublish(body)226			if ok && b.handler != nil {227				b.handler(topic, payload)228			}229		}230	}231}232233// Publish enqueues a QoS 0 PUBLISH from any goroutine. While the client234// is disconnected the queue does not exist and the publish is dropped,235// which is correct at QoS 0: a caller re-publishes its retained state236// from onConnect, so the broker holds the current value again within237// the reconnect.238func (b *Bus) Publish(topic string, payload []byte, retained bool) {239	b.mutex.Lock()240	out := b.out241	b.mutex.Unlock()242	if out == nil {243		return244	}245	select {246	case out <- encodePublish(topic, payload, retained):247	default:248	}249}250251// Subscribe remembers the filter and sends it if the client is252// connected. The remembered set is what runSession re-sends on every253// reconnect, so a subscription outlives the connection it was made on.254func (b *Bus) Subscribe(filter string) {255	b.mutex.Lock()256	b.filters[filter] = struct{}{}257	out := b.out258	frame := encodeSubscribe(b.nextPacketID(), filter)259	b.mutex.Unlock()260	if out == nil {261		return262	}263	select {264	case out <- frame:265	default:266	}267}268269// resubscribe sends every remembered filter on a fresh connection. The270// filters go out in sorted order so the frames a reconnect writes are271// the same every time, which keeps a test deterministic.272func (b *Bus) resubscribe(out chan<- []byte) {273	b.mutex.Lock()274	filters := make([]string, 0, len(b.filters))275	for filter := range b.filters {276		filters = append(filters, filter)277	}278	frames := make([][]byte, 0, len(filters))279	sort.Strings(filters)280	for _, filter := range filters {281		frames = append(frames, encodeSubscribe(b.nextPacketID(), filter))282	}283	b.mutex.Unlock()284	for _, frame := range frames {285		select {286		case out <- frame:287		default:288		}289	}290}291292// nextPacketID hands out the identifier a SUBSCRIBE carries and its293// SUBACK echoes. It skips zero, which the protocol reserves. The caller294// holds the mutex.295func (b *Bus) nextPacketID() uint16 {296	b.packetID++297	if b.packetID == 0 {298		b.packetID = 1299	}300	return b.packetID301}

1package main23// One receiver's control connection: the ASCII protocol a Denon speaks4// on port 23, the state the operator folds every line into, and the5// reconnect that keeps the socket honest.6// The protocol reference is https://assets.denon.com/documentmaster/uk/7// avr1713_avr1613_protocol_v860.pdf. The lines this file parses were8// also read from a live AVR-X1700H.910import (11	"bufio"12	"context"13	"net"14	"strings"15	"sync"16	"time"17)1819// The port a Denon answers control on, used when the declared address20// names none.21const denonPort = "23"2223// Commands and replies alike are ASCII terminated by a carriage return.24const denonTerminator = '\r'2526// The five queries the operator sends on every connect. Their answers27// are the whole of the status.28var denonQueries = []string{"PW?", "MV?", "MU?", "SI?", "MS?"}2930// The set commands this operator sends.31const (32	denonPowerOnCommand      = "PWON"33	denonMuteOnCommand       = "MUON"34	denonMuteOffCommand      = "MUOFF"35	denonVolumeCommandPrefix = "MV"36	denonInputCommandPrefix  = "SI"37)3839// The two power words the status carries.40const (41	powerOn      = "on"42	powerStandby = "standby"43)4445// The timings of one connection. The heartbeat is a query the receiver46// answers. Silence longer than the limit ends the session even though47// the socket is still open, because a half-open socket reads as live48// and swallows every write.49var (50	denonDialTimeout  = 5 * time.Second51	denonHeartbeat    = 30 * time.Second52	denonSilenceLimit = 75 * time.Second53	denonMinBackoff   = time.Second54	denonMaxBackoff   = 30 * time.Second55)5657// The write queue for one connection. A command that would overflow it58// is dropped, because the operator re-asserts nothing.59const denonQueueDepth = 326061// The fields a line can name, which is what a listener switches on.62const (63	denonPowerField     = "power"64	denonInputField     = "input"65	denonVolumeField    = "volume"66	denonVolumeMaxField = "volumeMax"67	denonMuteField      = "mute"68	denonSoundModeField = "soundMode"69	denonReachableField = "reachable"70)7172// denonState is what the receiver last said, in the receiver's own73// units, and whether the operator can still reach it.74type denonState struct {75	Power     string76	Input     string77	SoundMode string78	Mute      bool79	Volume    int80	VolumeMax int81	Reachable ConditionStatus82}8384// The state before the receiver has said anything.85func newDenonState() denonState {86	return denonState{87		Volume:    unknownHalves,88		VolumeMax: unknownHalves,89		Reachable: ConditionUnknown,90	}91}9293// denonEvent is one recognized line: the field it named and the whole94// state after it.95type denonEvent struct {96	Field string97	State denonState98}99100// applyDenonLine folds one line into the state and answers which field101// it named. A line this operator does not know is ignored, because a102// Denon volunteers dozens of settings nothing here reads.103func applyDenonLine(state denonState, line string) (denonState, string, bool) {104	switch {105	case line == "PWON":106		state.Power = powerOn107		return state, denonPowerField, true108	case line == "PWSTANDBY":109		state.Power = powerStandby110		return state, denonPowerField, true111	case line == denonMuteOnCommand:112		state.Mute = true113		return state, denonMuteField, true114	case line == denonMuteOffCommand:115		state.Mute = false116		return state, denonMuteField, true117	case strings.HasPrefix(line, "MVMAX "):118		halves, ok := parseHalfSteps(strings.TrimSpace(line[len("MVMAX "):]))119		if !ok {120			return state, "", false121		}122		state.VolumeMax = halves123		return state, denonVolumeMaxField, true124	case strings.HasPrefix(line, denonVolumeCommandPrefix):125		halves, ok := parseHalfSteps(line[len(denonVolumeCommandPrefix):])126		if !ok {127			return state, "", false128		}129		state.Volume = halves130		return state, denonVolumeField, true131	case strings.HasPrefix(line, denonInputCommandPrefix) && len(line) > 2:132		state.Input = line[2:]133		return state, denonInputField, true134	case strings.HasPrefix(line, "MS") && len(line) > 2:135		state.SoundMode = line[2:]136		return state, denonSoundModeField, true137	}138	return state, "", false139}140141// denonVolumeCommand is the set command for one half-step count.142func denonVolumeCommand(halves int) string {143	return denonVolumeCommandPrefix + halfStepDigits(halves)144}145146// denonMuteCommand is the set command for one mute state.147func denonMuteCommand(muted bool) string {148	if muted {149		return denonMuteOnCommand150	}151	return denonMuteOffCommand152}153154// denonInputCommand selects one input by the name the receiver carries155// for it.156func denonInputCommand(input string) string {157	return denonInputCommandPrefix + input158}159160// denonAddress fills in the control port when the declared address161// names none.162func denonAddress(address string) string {163	if _, _, err := net.SplitHostPort(address); err == nil {164		return address165	}166	return net.JoinHostPort(address, denonPort)167}168169// denonClient holds one receiver's connection, the state it reports,170// and the queue of commands waiting to go out. out is the current171// connection's queue, or nil while disconnected.172type denonClient struct {173	address  string174	listener func(denonEvent)175176	mutex sync.Mutex177	state denonState178	out   chan string179}180181func newDenonClient(address string, listener func(denonEvent)) *denonClient {182	return &denonClient{183		address:  denonAddress(address),184		listener: listener,185		state:    newDenonState(),186	}187}188189// State answers what the receiver last said, from any goroutine.190func (d *denonClient) State() denonState {191	d.mutex.Lock()192	defer d.mutex.Unlock()193	return d.state194}195196// Send queues one command. While the client is disconnected the queue197// does not exist and the command is dropped. Every command this198// operator sends is a one-shot, and a stale one sent after a reconnect199// would fight a hand on the equipment.200func (d *denonClient) Send(command string) {201	d.mutex.Lock()202	out := d.out203	d.mutex.Unlock()204	if out == nil {205		return206	}207	select {208	case out <- command:209	default:210	}211}212213// Run holds the connection open until ctx ends, and waits a backoff214// between sessions. A session that got an answer resets the backoff, so215// a receiver that was unplugged for an hour reconnects at once, while216// an address that never answers is retried more and more slowly.217func (d *denonClient) Run(ctx context.Context) {218	backoff := denonMinBackoff219	for ctx.Err() == nil {220		answered := d.runSession(ctx)221		d.record(ConditionFalse)222		if ctx.Err() != nil {223			return224		}225		if answered {226			backoff = denonMinBackoff227		}228		select {229		case <-ctx.Done():230			return231		case <-time.After(backoff):232		}233		if !answered {234			backoff *= 2235			if backoff > denonMaxBackoff {236				backoff = denonMaxBackoff237			}238		}239	}240}241242// runSession dials, asks the five queries, and reads until the receiver243// falls silent or the socket fails. It answers whether the receiver244// ever replied, which is what tells Run to reset the backoff.245func (d *denonClient) runSession(parent context.Context) (answered bool) {246	dialer := &net.Dialer{Timeout: denonDialTimeout}247	conn, err := dialer.DialContext(parent, "tcp", d.address)248	if err != nil {249		return false250	}251	defer conn.Close()252253	ctx, cancel := context.WithCancel(parent)254	defer cancel()255	// The reader blocks until the receiver writes or the socket closes.256	// Closing the socket is what unblocks a reader on a silent receiver.257	defer context.AfterFunc(ctx, func() { conn.Close() })()258259	out := make(chan string, denonQueueDepth)260	d.mutex.Lock()261	d.out = out262	d.mutex.Unlock()263	defer func() {264		d.mutex.Lock()265		d.out = nil266		d.mutex.Unlock()267	}()268269	var writing sync.WaitGroup270	writing.Add(1)271	go func() {272		defer writing.Done()273		defer cancel()274		d.writeLoop(ctx, conn, out)275	}()276277	for _, query := range denonQueries {278		d.Send(query)279	}280281	answered = d.readLoop(conn)282	cancel()283	writing.Wait()284	return answered285}286287// writeLoop is the one goroutine that writes the socket. It sends the288// heartbeat query whenever nothing else has gone out for a heartbeat's289// time.290func (d *denonClient) writeLoop(ctx context.Context, conn net.Conn, out <-chan string) {291	ticker := time.NewTicker(denonHeartbeat)292	defer ticker.Stop()293	last := time.Now()294	for {295		select {296		case <-ctx.Done():297			return298		case command := <-out:299			if _, err := conn.Write([]byte(command + string(denonTerminator))); err != nil {300				return301			}302			last = time.Now()303		case <-ticker.C:304			if time.Since(last) < denonHeartbeat {305				continue306			}307			if _, err := conn.Write([]byte("PW?" + string(denonTerminator))); err != nil {308				return309			}310			last = time.Now()311		}312	}313}314315// readLoop folds every line into the state until the receiver falls316// silent longer than the limit or the socket fails.317func (d *denonClient) readLoop(conn net.Conn) (answered bool) {318	reader := bufio.NewReader(conn)319	for {320		if err := conn.SetReadDeadline(time.Now().Add(denonSilenceLimit)); err != nil {321			return answered322		}323		line, err := reader.ReadString(denonTerminator)324		if err != nil {325			return answered326		}327		line = strings.Trim(line, "\r\n")328		if line == "" {329			continue330		}331		if d.fold(line) {332			answered = true333		}334	}335}336337// fold applies one line under the lock, hands the listener the state338// that came out, and answers whether the line was one this operator339// reads.340func (d *denonClient) fold(line string) bool {341	d.mutex.Lock()342	folded, field, known := applyDenonLine(d.state, line)343	if !known {344		d.mutex.Unlock()345		return false346	}347	// A line the operator reads is the answered round trip that Reachable348	// stands on.349	folded.Reachable = ConditionTrue350	d.state = folded351	d.mutex.Unlock()352353	d.notify(denonEvent{Field: field, State: folded})354	return true355}356357// record moves the reachability verdict and tells the listener, which358// is how a dropped connection reaches the status.359func (d *denonClient) record(status ConditionStatus) {360	d.mutex.Lock()361	if d.state.Reachable == status {362		d.mutex.Unlock()363		return364	}365	d.state.Reachable = status366	state := d.state367	d.mutex.Unlock()368369	d.notify(denonEvent{Field: denonReachableField, State: state})370}371372// notify runs on the reading goroutine, so a listener that blocks holds373// up every later line on the connection.374func (d *denonClient) notify(event denonEvent) {375	if d.listener != nil {376		d.listener(event)377	}378}

1package main23// The level path: the bus payload the room's level travels in, the4// owner mark that says who applies it, and the arithmetic between the5// bus scale and the receiver's own.67import (8	"encoding/json"9	"math"10	"strconv"11)1213// The bus scale runs 0 to 100, and 100 is as loud as the room ever14// goes.15const (16	minLevel = 017	maxLevel = 10018)1920// volumeState is the whole payload on a Player's volume topic, the21// shape every pod for the unit applies. It matches the media operator's22// own type.23type volumeState struct {24	Level int  `json:"level"`25	Muted bool `json:"muted"`26}2728// clamped holds the level inside the bus scale, so no arithmetic29// elsewhere bounds itself.30func (v volumeState) clamped() volumeState {31	if v.Level < minLevel {32		v.Level = minLevel33	}34	if v.Level > maxLevel {35		v.Level = maxLevel36	}37	return v38}3940// parseVolumeState reads one message off the topic. A payload that does41// not decode is no state at all.42func parseVolumeState(payload []byte) (volumeState, bool) {43	var state volumeState44	if err := json.Unmarshal(payload, &state); err != nil {45		return volumeState{}, false46	}47	return state.clamped(), true48}4950// marshalVolumeState encodes a state for the topic. The clamp runs here51// too, so nothing this operator publishes is out of range.52func marshalVolumeState(state volumeState) ([]byte, error) {53	return json.Marshal(state.clamped())54}5556// volumeOwner is the retained payload on <volumeTopic>/owner. While it57// stands, the named owner applies the level and every pod leaves mpv at58// unity.59type volumeOwner struct {60	Owner string `json:"owner"`61}6263// ownerTopic is where the mark for one volume topic sits.64func ownerTopic(volumeTopic string) string {65	return volumeTopic + "/owner"66}6768// ownerMark is the payload this operator publishes while it holds a69// receiver's session.70func ownerMark(receiver string) ([]byte, error) {71	return json.Marshal(volumeOwner{Owner: "receiver/" + receiver})72}7374// The receiver's scale is counted in half steps, because a Denon moves75// in halves and an integer count of them never rounds. unknownHalves is76// a value the receiver has not reported yet.77const unknownHalves = -17879// parseHalfSteps reads the digits a Denon sends. Two digits are whole80// steps and three digits are tenths, and the receiver only ever sends a81// tenth of five.82func parseHalfSteps(digits string) (int, bool) {83	value, err := strconv.Atoi(digits)84	if err != nil || value < 0 {85		return unknownHalves, false86	}87	switch len(digits) {88	case 2:89		return value * 2, true90	case 3:91		if value%5 != 0 {92			return unknownHalves, false93		}94		return value / 5, true95	}96	return unknownHalves, false97}9899// formatHalfSteps writes a count the way a person reads it, which is100// what the status carries.101func formatHalfSteps(halves int) string {102	if halves < 0 {103		return ""104	}105	if halves%2 == 0 {106		return strconv.Itoa(halves / 2)107	}108	return strconv.Itoa(halves/2) + ".5"109}110111// halfStepDigits writes the same count in the digits a set command112// carries: two for a whole step, three for a half.113func halfStepDigits(halves int) string {114	whole := halves / 2115	digits := strconv.Itoa(whole)116	if whole < 10 {117		digits = "0" + digits118	}119	if halves%2 == 1 {120		digits += "5"121	}122	return digits123}124125// halvesForLevel maps the bus scale onto the receiver's, so 100 is the126// ceiling the room is allowed. A ceiling that is not known yet cannot127// be mapped onto.128func halvesForLevel(level, maxHalves int) (int, bool) {129	if maxHalves <= 0 {130		return unknownHalves, false131	}132	if level < minLevel {133		level = minLevel134	}135	if level > maxLevel {136		level = maxLevel137	}138	return int(math.Round(float64(level) * float64(maxHalves) / maxLevel)), true139}140141// levelForHalves maps the receiver's scale back onto the bus against142// the same ceiling, which is what the session publishes after every143// move.144func levelForHalves(halves, maxHalves int) (int, bool) {145	if maxHalves <= 0 || halves < 0 {146		return 0, false147	}148	level := int(math.Round(float64(halves) * maxLevel / float64(maxHalves)))149	if level > maxLevel {150		level = maxLevel151	}152	return level, true153}154155// Where no step is declared, one press moves the receiver by one whole156// unit of its own scale, which is two half steps.157const defaultStepHalves = 2158159// halvesFromScale reads a figure a person wrote in the receiver's own160// scale as a count of half steps. A figure between two half steps is161// not one the receiver can take, so it goes to the nearest.162func halvesFromScale(value float64) int {163	return int(math.Round(value * 2))164}165166// The top of a Denon's own scale, 98, which no declared ceiling may167// exceed.168const denonScaleTop = 196169170// ceilingHalves is what 100 on the bus means. It is the ceiling a171// person declared and nothing else. The receiver's own MVMAX line is172// not a limit: one AVR reported 69.5, then 70.5, then 64.5 in one173// evening, so a ceiling read from it would move under the room.174func ceilingHalves(rule ReceiverVolume) int {175	stated := halvesFromScale(rule.Max)176	if stated <= 0 {177		return 0178	}179	if stated > denonScaleTop {180		return denonScaleTop181	}182	return stated183}184185// pressHalves is how far one press moves the receiver.186func pressHalves(rule ReceiverVolume) int {187	if stated := halvesFromScale(rule.Step); stated > 0 {188		return stated189	}190	return defaultStepHalves191}

1// equipment-operator drives the A/V receivers of a liken cluster and2// reports what each one says. It reaches each receiver over the3// network, applies the session a Player holds on it, and owns the4// room's level while that session stands.5package main67import "os"89// The operator's own environment. Each value is one the operator10// cannot derive, so the Deployment states it and this program reads11// it here.12const (13	// POD_NAMESPACE is the namespace the Deployment runs in. Nothing reads14	// it until the Service front lands.15	podNamespaceVariable = "POD_NAMESPACE"1617	// EQUIPMENT_BUS_ADDRESS is the broker a session's volume topic is read18	// from, as host:port.19	busAddressVariable = "EQUIPMENT_BUS_ADDRESS"20)2122// settings is the whole of the operator's configuration.23type settings struct {24	namespace  string25	busAddress string26}2728// readSettings takes the configuration from the environment alone.29// An unset variable reads as empty, and the caller decides what an30// empty value means, so a missing setting never stops the read.31func readSettings() settings {32	return settings{33		namespace:  os.Getenv(podNamespaceVariable),34		busAddress: os.Getenv(busAddressVariable),35	}36}3738func main() {39	operate()40}

1package main23// The MQTT 3.1.1 wire codec, written straight against the protocol4// the way apiclient.go writes the Kubernetes client. The broker is5// Mosquitto and the protocol is a published standard, so a client6// that speaks the few packets this operator needs costs less than a7// third-party library and its release cadence.8//9// This operator uses QoS 0 alone, because it republishes every retained10// payload it owns on each reconnect.1112import (13	"bufio"14	"fmt"15	"io"16)1718// The MQTT control packet types, in the high nibble of a fixed19// header's first byte. The low nibble carries per-type flags, which20// matter here only for a PUBLISH, where bit 0 is the retain flag.21const (22	mqttConnect   = 0x1023	mqttConnack   = 0x2024	mqttPublish   = 0x3025	mqttSubscribe = 0x8026	mqttSuback    = 0x9027	mqttPingreq   = 0xC028	mqttPingresp  = 0xD029)3031// The MQTT 3.1.1 protocol name and level. The name is the literal32// string "MQTT" and the level is 4, which together tell the broker33// which version of the protocol this client speaks.34const (35	mqttProtocolName  = "MQTT"36	mqttProtocolLevel = 0x0437)3839// The CONNECT flag bits this client sets. Clean session starts every40// connection with no server-side state, which is correct because the41// client re-subscribes and re-publishes on each connect. The will42// bits arrive only when the caller states a will. Username and43// password stay unset, because the in-cluster network is the trust44// boundary and the broker accepts the cluster's own pods.45const (46	connectCleanSession = 0x0247	connectWillFlag     = 0x0448	connectWillRetain   = 0x2049)5051// encodeRemainingLength writes the packet length in the variable-byte52// form the protocol uses: seven bits of length per byte, and the high53// bit set on every byte but the last. One byte covers up to 127, and54// four bytes cover the 268435455 the protocol allows.55func encodeRemainingLength(length int) []byte {56	var encoded []byte57	for {58		digit := byte(length % 128)59		length /= 12860		if length > 0 {61			digit |= 0x8062		}63		encoded = append(encoded, digit)64		if length == 0 {65			return encoded66		}67	}68}6970// decodeRemainingLength reads the variable-byte length back. It reads71// at most four bytes, because a fifth would exceed the protocol's72// limit and marks a stream that has lost frame alignment.73func decodeRemainingLength(reader io.ByteReader) (int, error) {74	length := 075	multiplier := 176	for count := 0; count < 4; count++ {77		digit, err := reader.ReadByte()78		if err != nil {79			return 0, err80		}81		length += int(digit&0x7F) * multiplier82		if digit&0x80 == 0 {83			return length, nil84		}85		multiplier *= 12886	}87	return 0, fmt.Errorf("mqtt: remaining length runs past four bytes")88}8990// appendString writes one length-prefixed UTF-8 string, the shape the91// protocol uses for a topic, a filter, and the client identifier: two92// bytes of length, most significant first, then the bytes.93func appendString(buffer []byte, value string) []byte {94	buffer = append(buffer, byte(len(value)>>8), byte(len(value)))95	return append(buffer, value...)96}9798// appendBytes writes one length-prefixed byte string, the shape a will99// payload takes.100func appendBytes(buffer []byte, value []byte) []byte {101	buffer = append(buffer, byte(len(value)>>8), byte(len(value)))102	return append(buffer, value...)103}104105// packet frames one control packet: a fixed-header first byte, then106// the remaining length, then the body. Every encode function ends107// here, so the length is computed once from the finished body.108func packet(first byte, body []byte) []byte {109	frame := make([]byte, 0, 2+len(body))110	frame = append(frame, first)111	frame = append(frame, encodeRemainingLength(len(body))...)112	return append(frame, body...)113}114115// encodeConnect builds the first packet the client sends. The body is116// the protocol name and level, one flags byte, the keepalive in117// seconds, and the payload: the client identifier and, when the caller118// states one, the will topic and payload. The client authenticates119// with nothing more than its identifier, because the broker accepts120// the cluster's own pods.121func encodeConnect(clientID string, keepalive uint16, will *busWill) []byte {122	flags := byte(connectCleanSession)123	if will != nil {124		flags |= connectWillFlag125		if will.Retained {126			flags |= connectWillRetain127		}128	}129130	var body []byte131	body = appendString(body, mqttProtocolName)132	body = append(body, mqttProtocolLevel, flags)133	body = append(body, byte(keepalive>>8), byte(keepalive))134	body = appendString(body, clientID)135	if will != nil {136		body = appendString(body, will.Topic)137		body = appendBytes(body, will.Payload)138	}139	return packet(mqttConnect, body)140}141142// encodePublish builds a QoS 0 PUBLISH. The retain flag is bit 0 of143// the first byte, and a retained publish tells the broker to hold this144// payload as the topic's last value and deliver it to every later145// subscriber. QoS 0 carries no packet identifier, so the body is the146// length-prefixed topic and then the raw payload.147func encodePublish(topic string, payload []byte, retained bool) []byte {148	first := byte(mqttPublish)149	if retained {150		first |= 0x01151	}152	body := appendString(nil, topic)153	body = append(body, payload...)154	return packet(first, body)155}156157// encodeSubscribe builds a SUBSCRIBE for one topic filter at QoS 0.158// The fixed header is 0x82, because bit 1 is reserved and must be set159// on a SUBSCRIBE. The body is the packet identifier the broker echoes160// in its SUBACK, then the length-prefixed filter and one byte of161// requested QoS.162func encodeSubscribe(packetID uint16, filter string) []byte {163	body := []byte{byte(packetID >> 8), byte(packetID)}164	body = appendString(body, filter)165	body = append(body, 0x00)166	return packet(mqttSubscribe|0x02, body)167}168169// encodePingreq builds the keepalive packet. It carries no body, so it170// is the two bytes 0xC0 0x00, and the broker answers with a PINGRESP.171func encodePingreq() []byte {172	return []byte{mqttPingreq, 0x00}173}174175// readPacket reads one whole control packet: the fixed-header first176// byte, the remaining length, and that many bytes of body. It returns177// the first byte so the caller reads both the packet type in the high178// nibble and the flags in the low nibble.179func readPacket(reader *bufio.Reader) (byte, []byte, error) {180	first, err := reader.ReadByte()181	if err != nil {182		return 0, nil, err183	}184	length, err := decodeRemainingLength(reader)185	if err != nil {186		return 0, nil, err187	}188	body := make([]byte, length)189	if _, err := io.ReadFull(reader, body); err != nil {190		return 0, nil, err191	}192	return first, body, nil193}194195// parseConnack reads the broker's answer to a CONNECT. The body is one196// byte of acknowledge flags and one byte of return code, and a return197// code other than zero is the broker refusing the connection.198func parseConnack(body []byte) error {199	if len(body) < 2 {200		return fmt.Errorf("mqtt: a CONNACK carried %d bytes, want 2", len(body))201	}202	if body[1] != 0x00 {203		return fmt.Errorf("mqtt: the broker refused the connection with code %d", body[1])204	}205	return nil206}207208// parseSuback reads the broker's answer to a SUBSCRIBE. The body is the209// echoed packet identifier and one return code per filter, and a210// return code of 0x80 is the broker refusing that subscription.211func parseSuback(body []byte) error {212	if len(body) < 3 {213		return fmt.Errorf("mqtt: a SUBACK carried %d bytes, want at least 3", len(body))214	}215	for _, code := range body[2:] {216		if code == 0x80 {217			return fmt.Errorf("mqtt: the broker refused a subscription")218		}219	}220	return nil221}222223// parsePublish reads an inbound PUBLISH body into its topic and224// payload. The client subscribes at QoS 0 alone, so an inbound publish225// carries no packet identifier and the payload begins right after the226// length-prefixed topic.227func parsePublish(body []byte) (topic string, payload []byte, ok bool) {228	if len(body) < 2 {229		return "", nil, false230	}231	topicLength := int(body[0])<<8 | int(body[1])232	if len(body) < 2+topicLength {233		return "", nil, false234	}235	return string(body[2 : 2+topicLength]), body[2+topicLength:], true236}

1package main23// The operator's loop: level-triggered, woken by a watch, with a ticker4// as the backstop. A pass reads the whole collection, so a lost event5// costs at most one tick and a restarted operator starts correct.6// One client per Receiver holds the receiver's connection for as long7// as the Receiver stands. A debounced writer folds the burst of lines8// that follows one change into a single status write.910import (11	"context"12	"fmt"13	"os"14	"sync"15	"sync/atomic"16	"time"17)1819// How often the loop reconciles with nothing to prompt it.20const backstopInterval = 30 * time.Second2122// How long a burst of lines is collected before one status write.23var statusDebounce = 250 * time.Millisecond2425// receiverUnit is one Receiver's running parts: the connection, the26// status writer, and the session that holds the level.27type receiverUnit struct {28	name       string29	address    string30	client     *Client31	busAddress string32	now        func() time.Time33	denon      *denonClient34	cancel     context.CancelFunc35	dirty      chan struct{}36	generation atomic.Int6437	// The ceiling and the step live here and not on the session, so an38	// edit to them reaches a standing session with no restart.39	volume atomic.Pointer[ReceiverVolume]4041	mutex   sync.Mutex42	session *session43	applied ReceiverStatus44	written bool45}4647// observe is where every line the receiver sends reaches the operator.48// It wakes the status writer, and it reaches the session that owns the49// level.50func (u *receiverUnit) observe(event denonEvent) {51	poke(u.dirty)52	u.mutex.Lock()53	held := u.session54	u.mutex.Unlock()55	if held != nil {56		held.observe(event)57	}58}5960// report writes the status a burst of lines settles on, one write per61// burst, and only when the write would change something.62func (u *receiverUnit) report(ctx context.Context) {63	for {64		select {65		case <-ctx.Done():66			return67		case <-u.dirty:68		}69		select {70		case <-ctx.Done():71			return72		case <-time.After(statusDebounce):73		}74		drainPokes(u.dirty)75		// A select answers a ready timer as readily as a ready context, so a76		// unit stopped inside the debounce is asked again here before it77		// writes.78		if ctx.Err() != nil {79			return80		}81		u.write()82	}83}8485func (u *receiverUnit) write() {86	status := buildReceiverStatus(u.denon.State(), u.generation.Load(), u.applied.Conditions, u.now())87	if u.written && sameStatus(status, u.applied) {88		return89	}90	if _, err := ApplyReceiverStatus(u.client, u.name, status); err != nil {91		fmt.Fprintf(os.Stderr, "writing the status of receiver %s: %v\n", u.name, err)92		return93	}94	u.applied, u.written = status, true95}9697// setSession starts, flips, replaces, or lifts the session. A session98// that has not changed is left alone, because power and input are one-99// shots the receiver answers once.100//101// A flip of either flag is not a change of session: it reaches the102// session that stands, which keeps its broker connection and its103// adopted level.104func (u *receiverUnit) setSession(ctx context.Context, spec *ReceiverSession) {105	u.mutex.Lock()106	held := u.session107	u.mutex.Unlock()108109	if held != nil && spec != nil && held.spec == spec.withoutFlags() {110		held.setFlags(spec.Active, spec.Awake)111		return112	}113	if held != nil {114		u.mutex.Lock()115		u.session = nil116		u.mutex.Unlock()117		held.stop()118	}119	if spec == nil {120		return121	}122	started := startSession(ctx, u.name, *spec, u.denon, u.busAddress, u.volumeRule)123	u.mutex.Lock()124	u.session = started125	u.mutex.Unlock()126}127128// setVolume records the ceiling and the step a person declared, which129// every press reads.130func (u *receiverUnit) setVolume(spec *ReceiverVolume) {131	rule := ReceiverVolume{}132	if spec != nil {133		rule = *spec134	}135	u.volume.Store(&rule)136}137138// volumeRule answers what the spec states now, so a press made after an139// edit is measured against the edited scale.140func (u *receiverUnit) volumeRule() ReceiverVolume {141	if held := u.volume.Load(); held != nil {142		return *held143	}144	return ReceiverVolume{}145}146147// stop lifts the session and closes the connection, which is what a148// deleted Receiver leaves behind.149func (u *receiverUnit) stop() {150	u.mutex.Lock()151	held := u.session152	u.session = nil153	u.mutex.Unlock()154	if held != nil {155		held.stop()156	}157	u.cancel()158}159160// controller holds what every pass needs and the units it runs.161type controller struct {162	client     *Client163	busAddress string164	wake       chan struct{}165	now        func() time.Time166	units      map[string]*receiverUnit167}168169func newController(client *Client, busAddress string) *controller {170	return &controller{171		client:     client,172		busAddress: busAddress,173		wake:       make(chan struct{}, 1),174		now:        time.Now,175		units:      map[string]*receiverUnit{},176	}177}178179// pass derives every unit from the collection as it stands now. It180// starts a client for a Receiver that names a protocol, moves a session181// that changed, and stops the client of a Receiver that is gone.182func (c *controller) pass(ctx context.Context) error {183	list, err := ListReceivers(c.client)184	if err != nil {185		return err186	}187188	live := map[string]bool{}189	for index := range list.Items {190		receiver := &list.Items[index]191		if receiver.Spec.Denon == nil {192			continue193		}194		live[receiver.Metadata.Name] = true195		c.reconcile(ctx, receiver)196	}197198	for name, unit := range c.units {199		if !live[name] {200			unit.stop()201			delete(c.units, name)202		}203	}204	return nil205}206207// reconcile brings one Receiver's unit up to its spec. An address that208// changed is a different receiver, so the unit is replaced and not209// redialled.210func (c *controller) reconcile(ctx context.Context, receiver *Receiver) {211	name := receiver.Metadata.Name212	unit, held := c.units[name]213	if held && unit.address != receiver.Spec.Denon.Address {214		unit.stop()215		delete(c.units, name)216		held = false217	}218	if !held {219		unit = c.start(ctx, receiver)220		c.units[name] = unit221	}222	unit.generation.Store(receiver.Metadata.Generation)223	unit.setVolume(receiver.Spec.Volume)224	unit.setSession(ctx, receiver.Spec.Session)225}226227func (c *controller) start(parent context.Context, receiver *Receiver) *receiverUnit {228	ctx, cancel := context.WithCancel(parent)229	unit := &receiverUnit{230		name:       receiver.Metadata.Name,231		address:    receiver.Spec.Denon.Address,232		client:     c.client,233		busAddress: c.busAddress,234		now:        c.now,235		cancel:     cancel,236		dirty:      make(chan struct{}, 1),237	}238	unit.setVolume(receiver.Spec.Volume)239	unit.denon = newDenonClient(receiver.Spec.Denon.Address, unit.observe)240	// The generation is stored before anything can write, so the first241	// status names the spec it was built from.242	unit.generation.Store(receiver.Metadata.Generation)243	go unit.denon.Run(ctx)244	go unit.report(ctx)245	// The first write says the operator holds the receiver and has not246	// reached it yet, before any line arrives.247	poke(unit.dirty)248	return unit249}250251// run reconciles once before any event arrives, then on every wake and252// every backstop tick, until ctx ends.253func (c *controller) run(ctx context.Context) {254	ticker := time.NewTicker(backstopInterval)255	defer ticker.Stop()256	for {257		if err := c.pass(ctx); err != nil {258			fmt.Fprintf(os.Stderr, "listing receivers: %v\n", err)259		}260		select {261		case <-ctx.Done():262			c.stopAll()263			return264		case <-c.wake:265		case <-ticker.C:266		}267	}268}269270func (c *controller) stopAll() {271	for name, unit := range c.units {272		unit.stop()273		delete(c.units, name)274	}275}276277// poke never blocks, and a wake channel buffers exactly one, because278// the pass that answers a wake reads the whole collection.279func poke(wake chan<- struct{}) {280	select {281	case wake <- struct{}{}:282	default:283	}284}285286// drainPokes clears the wakes a burst queued behind the one already287// taken.288func drainPokes(wake <-chan struct{}) {289	for {290		select {291		case <-wake:292		default:293			return294		}295	}296}297298// operate reads the configuration and hands it to serve. Every failure299// here ends the process, because the kubelet restarts the pod with300// backoff and the failure shows in kubectl instead of hiding in a retry301// loop.302func operate() {303	config := readSettings()304	if config.busAddress == "" {305		fmt.Fprintf(os.Stderr, "%s is unset; the Deployment must name the broker\n", busAddressVariable)306		os.Exit(1)307	}308309	client, err := InClusterClient()310	if err != nil {311		fmt.Fprintf(os.Stderr, "in-cluster config: %v\n", err)312		os.Exit(1)313	}314315	if err := serve(context.Background(), client, config.busAddress); err != nil {316		fmt.Fprintln(os.Stderr, err)317		os.Exit(1)318	}319}320321// serve proves the collection can be read, starts the watch from the322// version that first list carried, and runs the loop until ctx ends.323func serve(ctx context.Context, client *Client, busAddress string) error {324	list, err := ListReceivers(client)325	if err != nil {326		return fmt.Errorf("listing receivers: %w", err)327	}328329	operator := newController(client, busAddress)330	go watchReceivers(ctx, client, list.Metadata.ResourceVersion, operator.wake)331	operator.run(ctx)332	return nil333}

1package main23// The session a Player holds on a receiver: the one-shot power and4// input, the owner mark on the bus, the level the operator applies5// while the mark stands, and the knob turn it publishes back.67import (8	"context"9	"fmt"10	"os"11	"sync"12	"sync/atomic"13	"time"14)1516// How long the operator waits for the receiver to answer PWON before it17// selects the input anyway.18var sessionPowerWait = 10 * time.Second1920// How often the adopt looks again for what it needs. The spec that21// states the ceiling can land after the session starts, so an adopt22// that cannot map a level yet waits and does not give up.23var sessionAdoptRetry = 500 * time.Millisecond2425// How long a stop waits for the cleared owner mark to reach the broker26// before it closes the connection.27var sessionStopGrace = 200 * time.Millisecond2829// session is one live session: the connection to the broker, the level30// it last sent the receiver, and the marks that tell an echo of its own31// write from a hand on the equipment.32type session struct {33	receiver string34	spec     ReceiverSession35	denon    *denonClient36	bus      *Bus37	// The session's own context, held because a flip of either flag starts38	// the one-shots long after the session started, and they stop when it39	// does.40	ctx    context.Context41	cancel context.CancelFunc4243	// The two flags: whether a Play stands, and whether the room's screen44	// is awake. They gate the one-shots and nothing else. The level path45	// runs whatever they say.46	active atomic.Bool47	awake  atomic.Bool4849	// The gate the input selection waits on, closed when the receiver50	// reports itself on. It is armed again for each selection, so a second51	// Play waits for its own power-on.52	powerMutex sync.Mutex53	powered    chan struct{}5455	reachedOnce sync.Once56	reached     chan struct{}5758	completeOnce sync.Once59	complete     chan struct{}6061	connectedOnce sync.Once62	connected     chan struct{}6364	// scale is read on every press and never held, so a spec edit reaches65	// a standing session with no restart.66	scale func() ReceiverVolume6768	mutex      sync.Mutex69	latest     volumeState70	haveLatest bool7172	awaiting     volumeState73	haveAwaiting bool74	adopted      bool75}7677// startSession opens the session's own broker connection and claims the78// level with a retained owner mark.79//80// Power and input go out once for a session that starts with either81// flag on, and once only when both are on at the start. A session that82// starts with both off owns the level and sends the equipment nothing.83func startSession(ctx context.Context, receiver string, spec ReceiverSession, denon *denonClient, busAddress string, scale func() ReceiverVolume) *session {84	ctx, cancel := context.WithCancel(ctx)85	s := &session{86		receiver:  receiver,87		spec:      spec.withoutFlags(),88		denon:     denon,89		ctx:       ctx,90		cancel:    cancel,91		scale:     scale,92		powered:   make(chan struct{}),93		reached:   make(chan struct{}),94		complete:  make(chan struct{}),95		connected: make(chan struct{}),96	}97	s.mark(denon.State())9899	// The will clears the mark, so an operator that dies hands the level100	// back to the pods that were leaving it alone.101	will := &busWill{Topic: ownerTopic(spec.VolumeTopic), Retained: true}102	s.bus = newBus(busAddress, "equipment-operator-"+receiver, will, s.claim, s.receive)103	s.bus.Subscribe(spec.VolumeTopic)104	go s.bus.Run(ctx)105	go s.adopt(ctx)106	s.setFlags(spec.Active, spec.Awake)107	return s108}109110// setFlags takes both flags as the media operator wrote them: active111// when a Play starts or ends, awake when the room's screen wakes or112// sleeps. Either one turning on runs the one-shots once, and both113// turning on in one write runs them once and not twice. A screen that114// wakes under a standing Play selects the input again, which is what a115// person who reached for the receiver's own power button needs.116func (s *session) setFlags(active, awake bool) {117	played := s.raise(&s.active, active)118	woke := s.raise(&s.awake, awake)119	if played || woke {120		go s.selectInput(s.ctx)121	}122}123124// raise stores one flag and answers whether this is the false to true125// that runs the one-shots. True to false sends nothing, because the126// room may still be listening.127func (s *session) raise(flag *atomic.Bool, on bool) bool {128	if !on {129		flag.Store(false)130		return false131	}132	return flag.CompareAndSwap(false, true)133}134135// stop clears the owner mark, waits for it to reach the broker, and136// closes the connection. It never powers the receiver off, because the137// room may still be listening to something else.138func (s *session) stop() {139	s.publishOwner(nil)140	time.Sleep(sessionStopGrace)141	s.cancel()142}143144// claim publishes the mark on every fresh broker session, because a145// broker that restarted holds none of it.146func (s *session) claim(*Bus) {147	mark, err := ownerMark(s.receiver)148	if err != nil {149		fmt.Fprintf(os.Stderr, "marking the owner of %s: %v\n", s.spec.VolumeTopic, err)150		return151	}152	s.publishOwner(mark)153	s.connectedOnce.Do(func() { close(s.connected) })154}155156func (s *session) publishOwner(payload []byte) {157	s.bus.Publish(ownerTopic(s.spec.VolumeTopic), payload, true)158}159160// receive reads one message off the topic. A message that differs from161// the state the session holds is a press, and the session moves the162// receiver one step in its direction.163func (s *session) receive(topic string, payload []byte) {164	if topic != s.spec.VolumeTopic {165		return166	}167	state, ok := parseVolumeState(payload)168	if !ok {169		return170	}171	s.mutex.Lock()172	// The broker delivers a client's own publish back to it, and it sends173	// the topic's old retained state ahead of it. So the session's own174	// adopt message coming back is the line between the level a previous175	// session left and a press meant for this one. Everything before that176	// line was written for mpv's scale and moves nothing.177	if !s.adopted {178		if s.haveAwaiting && s.awaiting == state {179			s.latest, s.haveLatest, s.adopted = state, true, true180		}181		s.mutex.Unlock()182		return183	}184	// A state the session already holds is its own message coming back.185	// Applying it again would map the level onto a half step the equipment186	// is not on.187	if s.haveLatest && s.latest == state {188		s.mutex.Unlock()189		return190	}191	previous := s.latest192	s.latest, s.haveLatest = state, true193	s.mutex.Unlock()194195	s.press(previous, state)196}197198// press reads the message as a direction and moves the receiver one199// step from where it actually stands, never to a level mapped through200// two scales. Mute is absolute.201func (s *session) press(previous, state volumeState) {202	reading := s.denon.State()203	sent := false204	if state.Muted != reading.Mute {205		s.denon.Send(denonMuteCommand(state.Muted))206		sent = true207	}208	if state.Level != previous.Level {209		if target, moves := s.nextPosition(reading, state.Level > previous.Level); moves {210			s.denon.Send(denonVolumeCommand(target))211			sent = true212		}213	}214	// A press the receiver answers is reported when its answer arrives.215	// One that moves nothing has no answer coming, so the position goes216	// back to the topic now.217	if !sent {218		s.report(reading)219	}220}221222// nextPosition answers where one press puts the receiver, and whether223// it moves at all.224func (s *session) nextPosition(reading denonState, up bool) (int, bool) {225	ceiling := ceilingHalves(s.scale())226	if ceiling <= 0 || reading.Volume < 0 {227		return 0, false228	}229	// A hand can leave the receiver above the ceiling. A press up then230	// moves nothing, and never drops the receiver to the ceiling.231	if up && reading.Volume >= ceiling {232		return 0, false233	}234	// The ceiling bounds the way up and never the way down, so a receiver235	// above it steps down one press at a time.236	step := pressHalves(s.scale())237	if !up {238		target := max(reading.Volume-step, 0)239		return target, target != reading.Volume240	}241	target := min(reading.Volume+step, ceiling)242	return target, target != reading.Volume243}244245// report puts where the receiver actually stands back on the topic, so246// the sidecar's next press counts from a value that matches the247// equipment.248func (s *session) report(reading denonState) {249	level, ok := levelForHalves(reading.Volume, ceilingHalves(s.scale()))250	if !ok {251		return252	}253	position := volumeState{Level: level, Muted: reading.Mute}.clamped()254	payload, err := marshalVolumeState(position)255	if err != nil {256		fmt.Fprintf(os.Stderr, "publishing the level of %s: %v\n", s.spec.VolumeTopic, err)257		return258	}259260	s.mutex.Lock()261	defer s.mutex.Unlock()262	if s.haveLatest && s.latest == position {263		return264	}265	s.bus.Publish(s.spec.VolumeTopic, payload, true)266	s.latest, s.haveLatest = position, true267}268269// observe is the session's half of every line the receiver sends. It270// releases the waits the one-shots stand on, and it reports a position271// the operator did not ask for, such as a knob turn.272func (s *session) observe(event denonEvent) {273	s.mark(event.State)274	switch event.Field {275	case denonVolumeField, denonMuteField:276		s.report(event.State)277	}278}279280// mark releases the three waits a session stands on: the connection the281// commands go out over, the power the input selection follows, and the282// volume reading the adopt needs.283func (s *session) mark(state denonState) {284	if state.Reachable == ConditionTrue {285		s.reachedOnce.Do(func() { close(s.reached) })286	}287	if state.Power == powerOn {288		s.notePower()289	}290	if state.Reachable == ConditionTrue && state.Volume != unknownHalves {291		s.completeOnce.Do(func() { close(s.complete) })292	}293}294295// notePower releases whoever waits for the receiver to come on.296func (s *session) notePower() {297	s.powerMutex.Lock()298	defer s.powerMutex.Unlock()299	select {300	case <-s.powered:301	default:302		close(s.powered)303	}304}305306// armPower answers the gate that closes when the receiver next reports307// itself on. A selection arms it before it reads the power, so a308// receiver that answers in between still releases the wait.309func (s *session) armPower() <-chan struct{} {310	s.powerMutex.Lock()311	defer s.powerMutex.Unlock()312	s.powered = make(chan struct{})313	return s.powered314}315316// adopt is what makes the session willing to apply a level. The topic317// holds whatever the last session left on it, in mpv's scale and not318// this receiver's. So the session publishes where the equipment already319// stands and takes that as the state it holds. Until that message comes320// back, every message on the topic is history and moves nothing.321func (s *session) adopt(ctx context.Context) {322	select {323	case <-ctx.Done():324		return325	case <-s.complete:326	}327	select {328	case <-ctx.Done():329		return330	case <-s.connected:331	}332333	ticker := time.NewTicker(sessionAdoptRetry)334	defer ticker.Stop()335	for !s.publishPosition() {336		select {337		case <-ctx.Done():338			return339		case <-ticker.C:340		}341	}342}343344// publishPosition puts the receiver's position on the topic and answers345// whether it went out. Everything that can stop it is something the346// next try may have: a ceiling nobody had declared yet, or a volume the347// receiver has not reported.348func (s *session) publishPosition() bool {349	state := s.denon.State()350	level, ok := levelForHalves(state.Volume, ceilingHalves(s.scale()))351	if !ok {352		return false353	}354	held := volumeState{Level: level, Muted: state.Mute}.clamped()355	payload, err := marshalVolumeState(held)356	if err != nil {357		fmt.Fprintf(os.Stderr, "adopting the level of %s: %v\n", s.spec.VolumeTopic, err)358		return false359	}360361	s.mutex.Lock()362	defer s.mutex.Unlock()363	s.bus.Publish(s.spec.VolumeTopic, payload, true)364	s.awaiting, s.haveAwaiting = held, true365	return true366}367368// selectInput powers the receiver on, waits for it to say so, and369// selects the input once for whoever asked.370//371// It runs once per flip of either flag, and never re-asserts inside372// one: a hand on the equipment outranks the cluster.373func (s *session) selectInput(ctx context.Context) {374	// A command sent before the connection is open is dropped, and a one-375	// shot is never re-asserted, so the wait for the connection is what376	// makes the one-shot land. The session's own lifetime is the bound: a377	// receiver that never answers has nothing to select.378	select {379	case <-ctx.Done():380		return381	case <-s.reached:382	}383	powered := s.armPower()384	if s.denon.State().Power != powerOn {385		s.denon.Send(denonPowerOnCommand)386		select {387		case <-ctx.Done():388			return389		case <-powered:390		case <-time.After(sessionPowerWait):391		}392	}393	if ctx.Err() != nil {394		return395	}396	s.denon.Send(denonInputCommand(s.spec.Input))397}

1package main23// The status one receiver reports: what it last said, in its own units,4// and the one condition that says whether the operator can still reach5// it.67import "time"89// The one condition this operator reports, and the reason for each10// verdict.11const (12	reachableConditionType = "Reachable"13	reasonConnected        = "Connected"14	reasonUnreachable      = "Unreachable"15	reasonConnecting       = "Connecting"16)1718// reachableWords is the reason and the message each verdict carries.19func reachableWords(status ConditionStatus) (reason, message string) {20	switch status {21	case ConditionTrue:22		return reasonConnected, "the receiver answered"23	case ConditionFalse:24		return reasonUnreachable, "the receiver did not answer"25	}26	return reasonConnecting, "the operator has not reached the receiver yet"27}2829// timestamp writes a moment the way the API server holds one.30func timestamp(at time.Time) string {31	return at.UTC().Format(time.RFC3339)32}3334// reachable builds the condition. It keeps the moment the verdict last35// changed, so the stamp moves only when the verdict flips.36func reachable(status ConditionStatus, generation int64, previous []Condition, now time.Time) Condition {37	reason, message := reachableWords(status)38	condition := Condition{39		Type:               reachableConditionType,40		Status:             status,41		ObservedGeneration: generation,42		Reason:             reason,43		Message:            message,44		LastTransitionTime: timestamp(now),45	}46	for _, held := range previous {47		if held.Type == reachableConditionType && held.Status == status && held.LastTransitionTime != "" {48			condition.LastTransitionTime = held.LastTransitionTime49		}50	}51	return condition52}5354// buildReceiverStatus is the whole status one receiver's state makes,55// in the receiver's own units. status.service stays empty until the56// operator makes the Service front.57func buildReceiverStatus(state denonState, generation int64, previous []Condition, now time.Time) ReceiverStatus {58	return ReceiverStatus{59		Power:      state.Power,60		Input:      state.Input,61		Volume:     formatHalfSteps(state.Volume),62		VolumeMax:  formatHalfSteps(state.VolumeMax),63		Mute:       state.Mute,64		SoundMode:  state.SoundMode,65		Conditions: []Condition{reachable(state.Reachable, generation, previous, now)},66	}67}6869// sameStatus answers whether a write would change anything.70func sameStatus(a, b ReceiverStatus) bool {71	if a.Power != b.Power || a.Input != b.Input || a.Volume != b.Volume ||72		a.VolumeMax != b.VolumeMax || a.Mute != b.Mute ||73		a.SoundMode != b.SoundMode || a.Service != b.Service ||74		len(a.Conditions) != len(b.Conditions) {75		return false76	}77	for index := range a.Conditions {78		if a.Conditions[index] != b.Conditions[index] {79			return false80		}81	}82	return true83}

1package main23// The watch is an ordinary GET whose response never ends. The API4// server holds the connection open and writes one JSON event per5// change. The event carries no object to the loop, because every pass6// re-lists, so an event is only a wake.78import (9	"context"10	"encoding/json"11	"fmt"12	"net/http"13	"os"14	"time"15)1617// How long a watch waits before it re-lists after a dropped stream.18var watchRetryPause = 2 * time.Second1920// watchReceivers wakes the loop on every change and resumes each stream21// from a resourceVersion, so no change is missed between reconnects. A22// dropped stream and a 410 Gone recover the same way: list the23// collection, wake the loop, and watch again from the list's own24// version.25func watchReceivers(ctx context.Context, client *Client, resourceVersion string, wake chan<- struct{}) {26	for ctx.Err() == nil {27		// The request carries the context, because a read of a stream that28		// never ends blocks until the far end writes, and closing the body29		// from elsewhere waits on that same read.30		resp, err := WatchReceivers(ctx, client, resourceVersion)31		if err == nil && resp.StatusCode == http.StatusOK {32			resourceVersion = readWatchStream(resp, resourceVersion, wake)33		}34		if resp != nil {35			drain(resp.Body)36		}3738		select {39		case <-ctx.Done():40			return41		case <-time.After(watchRetryPause):42		}43		list, err := ListReceivers(client)44		if err != nil {45			fmt.Fprintf(os.Stderr, "listing receivers to resume the watch: %v\n", err)46			continue47		}48		resourceVersion = list.Metadata.ResourceVersion49		poke(wake)50	}51}5253// readWatchStream reads one connection's events. The returned version54// is where the next watch resumes.55func readWatchStream(resp *http.Response, resourceVersion string, wake chan<- struct{}) string {56	decoder := json.NewDecoder(resp.Body)57	for {58		var event struct {59			Type   string `json:"type"`60			Object struct {61				Metadata ObjectMeta `json:"metadata"`62			} `json:"object"`63		}64		if err := decoder.Decode(&event); err != nil {65			return resourceVersion66		}67		if event.Type == "ERROR" {68			return resourceVersion69		}70		if event.Object.Metadata.ResourceVersion != "" {71			resourceVersion = event.Object.Metadata.ResourceVersion72		}73		if event.Type == "BOOKMARK" {74			continue75		}76		poke(wake)77	}78}