internal/server/api/notifier.go

Ref: Size: 1.1 KiB History
package api

import "sync"

// notifier is a tiny fan-out broadcaster: SSE subscribers get a (coalesced)
// wake-up whenever desired state changes. It carries no payload — woken
// subscribers re-read the current snapshot — so a burst of mutations collapses
// into at most one pending wake per subscriber.
type notifier struct {
	mu   sync.Mutex
	subs map[chan struct{}]struct{}
}

func newNotifier() *notifier {
	return &notifier{subs: make(map[chan struct{}]struct{})}
}

// subscribe returns a wake channel and an unsubscribe func. The channel has
// buffer 1 so notify never blocks and repeated notifies coalesce.
func (n *notifier) subscribe() (<-chan struct{}, func()) {
	ch := make(chan struct{}, 1)
	n.mu.Lock()
	n.subs[ch] = struct{}{}
	n.mu.Unlock()
	return ch, func() {
		n.mu.Lock()
		delete(n.subs, ch)
		n.mu.Unlock()
	}
}

// notify wakes all subscribers without blocking.
func (n *notifier) notify() {
	n.mu.Lock()
	defer n.mu.Unlock()
	for ch := range n.subs {
		select {
		case ch <- struct{}{}:
		default: // already has a pending wake; coalesce
		}
	}
}