128 lines
4.4 KiB
Go
Executable File
128 lines
4.4 KiB
Go
Executable File
// Package eventbus is the typed pub/sub that connects the Cast pipeline,
|
|
// the CLI, and the Coven Mirror WebUI.
|
|
//
|
|
// Every cast is identified by a taskID (a short UUID). The pipeline
|
|
// publishes log lines, progress events, and completion/failure events to
|
|
// the bus under that topic. The CLI subscribes to the same topic and
|
|
// prints to stdout; the WebUI's /api/v1/stream/{id} WebSocket handler
|
|
// subscribes and forwards to the browser.
|
|
//
|
|
// This is the glue that makes the three interfaces (CLI / TUI / WebUI)
|
|
// show the same real-time truth.
|
|
package eventbus
|
|
|
|
import (
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// EventType classifies an event.
|
|
type EventType string
|
|
|
|
const (
|
|
EventLog EventType = "log" // a build log line
|
|
EventProgress EventType = "progress" // {done, total}
|
|
EventPhase EventType = "phase" // "summoning" | "unpacking" | "building" | "committing"
|
|
EventComplete EventType = "complete" // success
|
|
EventFailed EventType = "failed" // error
|
|
EventAlarm EventType = "alarm" // warding alert
|
|
)
|
|
|
|
// Event is one published message.
|
|
type Event struct {
|
|
Topic string `json:"topic"`
|
|
Type EventType `json:"type"`
|
|
Data string `json:"data"`
|
|
Time time.Time `json:"time"`
|
|
Done int `json:"done,omitempty"`
|
|
Total int `json:"total,omitempty"`
|
|
}
|
|
|
|
// DefaultSubscriberBufSize is the buffer capacity for each subscriber channel.
|
|
// Overflow events are dropped (non-blocking publish) to avoid blocking
|
|
// the Cast pipeline on slow WebSocket clients.
|
|
const DefaultSubscriberBufSize = 64
|
|
|
|
// Bus is the in-memory pub/sub. Safe for concurrent publishers/subscribers.
|
|
type Bus struct {
|
|
mu sync.RWMutex
|
|
subs map[string][]chan Event
|
|
}
|
|
|
|
// New returns an empty Bus.
|
|
func New() *Bus {
|
|
return &Bus{subs: make(map[string][]chan Event)}
|
|
}
|
|
|
|
// Publish broadcasts an event to every subscriber of `topic`.
|
|
// Non-blocking: if a subscriber's buffer is full the event is dropped
|
|
// (we never block the Cast pipeline on a slow WebSocket).
|
|
func (b *Bus) Publish(topic string, ev Event) {
|
|
ev.Topic = topic
|
|
if ev.Time.IsZero() {
|
|
ev.Time = time.Now()
|
|
}
|
|
b.mu.RLock()
|
|
subs := b.subs[topic]
|
|
channels := make([]chan Event, len(subs))
|
|
copy(channels, subs)
|
|
b.mu.RUnlock()
|
|
for _, ch := range channels {
|
|
select {
|
|
case ch <- ev:
|
|
default:
|
|
// Slow consumer — event dropped. This is intentional: we never
|
|
// block the Cast pipeline on a lagging WebSocket.
|
|
log.Printf("eventbus: dropped event type=%s topic=%s (slow consumer)", ev.Type, ev.Topic)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Subscribe registers a receiver for `topic`. Returns the channel and an
|
|
// Unsubscribe func. The channel is buffered (DefaultSubscriberBufSize events);
|
|
// overflow is dropped via Publish's non-blocking send.
|
|
func (b *Bus) Subscribe(topic string) (<-chan Event, func()) {
|
|
ch := make(chan Event, DefaultSubscriberBufSize)
|
|
b.mu.Lock()
|
|
b.subs[topic] = append(b.subs[topic], ch)
|
|
b.mu.Unlock()
|
|
return ch, func() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
subs := b.subs[topic]
|
|
for i, c := range subs {
|
|
if c == ch {
|
|
b.subs[topic] = append(subs[:i], subs[i+1:]...)
|
|
close(c)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Log is a convenience helper for publishing a single log line.
|
|
func (b *Bus) Log(topic, line string) {
|
|
b.Publish(topic, Event{Type: EventLog, Data: line})
|
|
}
|
|
|
|
// Phase announces a pipeline phase change.
|
|
func (b *Bus) Phase(topic, phase string) {
|
|
b.Publish(topic, Event{Type: EventPhase, Data: phase})
|
|
}
|
|
|
|
// Progress announces a step completion.
|
|
func (b *Bus) Progress(topic string, done, total int) {
|
|
b.Publish(topic, Event{Type: EventProgress, Done: done, Total: total})
|
|
}
|
|
|
|
// Complete signals successful finish.
|
|
func (b *Bus) Complete(topic, essenceID string) {
|
|
b.Publish(topic, Event{Type: EventComplete, Data: essenceID})
|
|
}
|
|
|
|
// Failed signals an error.
|
|
func (b *Bus) Failed(topic, err string) {
|
|
b.Publish(topic, Event{Type: EventFailed, Data: err})
|
|
}
|