internal/server/hub/hub.go

Ref: main v0.0.5   Size: 1.5 KiB   History

// Package hub wakes per-host QUIC streams when desired state changes.
package hub

import "sync"

type Hub struct {
	mu sync.Mutex
	m  map[string]chan struct{}
}

func New() *Hub { return &Hub{m: map[string]chan struct{}{}} }

// Subscribe returns a channel that receives a poke whenever desired state
// changes for hostID, and a cancel function to deregister. A second Subscribe
// for the same host supersedes the first (last-writer-wins for reconnecting
// agents); the displaced channel is closed so any orphaned reader unblocks.
// The returned channel is closed when the subscription is cancelled or superseded,
// so callers using "for range" on the channel will exit naturally.
func (h *Hub) Subscribe(hostID string) (<-chan struct{}, func()) {
	h.mu.Lock()
	defer h.mu.Unlock()
	ch := make(chan struct{}, 1)
	if old := h.m[hostID]; old != nil {
		close(old)
	}
	h.m[hostID] = ch
	return ch, func() {
		h.mu.Lock()
		defer h.mu.Unlock()
		if h.m[hostID] == ch {
			delete(h.m, hostID)
			close(ch) // unblocks any goroutine ranging over this channel
		}
	}
}

func (h *Hub) Poke(hostID string) {
	// The send happens under the lock: Subscribe closes displaced channels,
	// and a send racing that close would panic. The send is non-blocking
	// (buffered-1 + default), so holding the lock here cannot deadlock.
	h.mu.Lock()
	defer h.mu.Unlock()
	ch := h.m[hostID]
	if ch == nil {
		return
	}
	select {
	case ch <- struct{}{}:
	default: // already pending; level-triggered consumers re-read full state anyway
	}
}