80 lines
1.5 KiB
Go
80 lines
1.5 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"sync"
|
|
)
|
|
|
|
const defaultBuffer = 64
|
|
|
|
// Publisher is the site-facing write side of the bus.
|
|
type Publisher interface {
|
|
Publish(ctx context.Context, ev any)
|
|
}
|
|
|
|
// Bus is an in-process pub/sub with one worker and a bounded queue.
|
|
type Bus struct {
|
|
ch chan any
|
|
mu sync.Mutex
|
|
subs []func(context.Context, any)
|
|
closed sync.Once
|
|
}
|
|
|
|
// New starts a worker that delivers events to subscribers in publish order.
|
|
func New() *Bus {
|
|
return newBus(defaultBuffer, true)
|
|
}
|
|
|
|
func newBus(buffer int, start bool) *Bus {
|
|
if buffer < 1 {
|
|
buffer = 1
|
|
}
|
|
b := &Bus{ch: make(chan any, buffer)}
|
|
if start {
|
|
go b.loop()
|
|
}
|
|
return b
|
|
}
|
|
|
|
// Publish enqueues ev. It never blocks the caller; a full buffer is dropped.
|
|
func (b *Bus) Publish(_ context.Context, ev any) {
|
|
if b == nil {
|
|
return
|
|
}
|
|
select {
|
|
case b.ch <- ev:
|
|
default:
|
|
log.Printf("events: dropped %T", ev)
|
|
}
|
|
}
|
|
|
|
// Subscribe adds a handler. Handlers run serially on the worker.
|
|
func (b *Bus) Subscribe(fn func(context.Context, any)) {
|
|
if b == nil || fn == nil {
|
|
return
|
|
}
|
|
b.mu.Lock()
|
|
b.subs = append(b.subs, fn)
|
|
b.mu.Unlock()
|
|
}
|
|
|
|
func (b *Bus) loop() {
|
|
for ev := range b.ch {
|
|
b.mu.Lock()
|
|
subs := append([]func(context.Context, any){}, b.subs...)
|
|
b.mu.Unlock()
|
|
for _, fn := range subs {
|
|
fn(context.Background(), ev)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close stops the worker. Safe to call more than once.
|
|
func (b *Bus) Close() {
|
|
if b == nil {
|
|
return
|
|
}
|
|
b.closed.Do(func() { close(b.ch) })
|
|
}
|