65 lines
1.6 KiB
Go
65 lines
1.6 KiB
Go
// Package events fans out "something changed" signals from the agent hub to
|
|
// open browser streams.
|
|
package events
|
|
|
|
import (
|
|
"strconv"
|
|
"sync"
|
|
)
|
|
|
|
// Topic names what changed. Subscribers refetch that part of the UI.
|
|
type Topic string
|
|
|
|
// Topics.
|
|
const (
|
|
Services Topic = "services"
|
|
Agents Topic = "agents"
|
|
Jobs Topic = "jobs" // a job was created or changed status
|
|
)
|
|
|
|
// JobTopic is signalled when job id gets new output or changes status. The
|
|
// name avoids ":" so it is safe inside hx-trigger="sse:job-42".
|
|
func JobTopic(id int64) Topic { return Topic("job-" + strconv.FormatInt(id, 10)) }
|
|
|
|
// Broker is an in-process publish/subscribe hub.
|
|
type Broker struct {
|
|
mu sync.Mutex
|
|
subs map[chan Topic]struct{}
|
|
}
|
|
|
|
// NewBroker returns an empty Broker.
|
|
func NewBroker() *Broker {
|
|
return &Broker{subs: make(map[chan Topic]struct{})}
|
|
}
|
|
|
|
// Subscribe returns a channel of topics and a cancel function that must be
|
|
// called when the subscriber goes away. Cancel is safe to call twice.
|
|
func (b *Broker) Subscribe() (<-chan Topic, func()) {
|
|
ch := make(chan Topic, 16)
|
|
b.mu.Lock()
|
|
b.subs[ch] = struct{}{}
|
|
b.mu.Unlock()
|
|
var once sync.Once
|
|
return ch, func() {
|
|
once.Do(func() {
|
|
b.mu.Lock()
|
|
delete(b.subs, ch)
|
|
b.mu.Unlock()
|
|
})
|
|
}
|
|
}
|
|
|
|
// Publish signals t to every subscriber. A subscriber whose buffer is full is
|
|
// skipped rather than blocking the agent hub; it already has signals queued
|
|
// and will refetch current state when it processes them.
|
|
func (b *Broker) Publish(t Topic) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for ch := range b.subs {
|
|
select {
|
|
case ch <- t:
|
|
default:
|
|
}
|
|
}
|
|
}
|