117 lines
2.7 KiB
Go
117 lines
2.7 KiB
Go
package events
|
|
|
|
import (
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
// Bus is a labeled publish/subscribe event bus.
|
|
// Handlers subscribe to patterns (exact or trailing wildcard).
|
|
// Publishing fans out to all matching subscribers.
|
|
type Bus struct {
|
|
mu sync.RWMutex
|
|
subs map[string][]*subscription
|
|
seq uint64 // subscription ID counter
|
|
}
|
|
|
|
type subscription struct {
|
|
id uint64
|
|
pattern string
|
|
handler Handler
|
|
}
|
|
|
|
// NewBus creates a new event bus.
|
|
func NewBus() *Bus {
|
|
return &Bus{
|
|
subs: make(map[string][]*subscription),
|
|
}
|
|
}
|
|
|
|
// Subscribe registers a handler for a label pattern.
|
|
// Returns an unsubscribe function.
|
|
//
|
|
// Patterns:
|
|
//
|
|
// "chat.message.abc123" — exact match
|
|
// "chat.message.*" — wildcard: matches chat.message.{anything}
|
|
// "chat.*" — wildcard: matches chat.{anything}
|
|
// "*" — matches all events
|
|
func (b *Bus) Subscribe(pattern string, handler Handler) func() {
|
|
b.mu.Lock()
|
|
b.seq++
|
|
sub := &subscription{id: b.seq, pattern: pattern, handler: handler}
|
|
b.subs[pattern] = append(b.subs[pattern], sub)
|
|
b.mu.Unlock()
|
|
|
|
return func() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
subs := b.subs[pattern]
|
|
for i, s := range subs {
|
|
if s.id == sub.id {
|
|
b.subs[pattern] = append(subs[:i], subs[i+1:]...)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Publish dispatches an event to all matching subscribers.
|
|
// Handlers are called synchronously in subscription order.
|
|
// Use PublishAsync for non-blocking dispatch.
|
|
func (b *Bus) Publish(event Event) {
|
|
b.mu.RLock()
|
|
var matched []Handler
|
|
for pattern, subs := range b.subs {
|
|
if match(event.Label, pattern) {
|
|
for _, s := range subs {
|
|
matched = append(matched, s.handler)
|
|
}
|
|
}
|
|
}
|
|
b.mu.RUnlock()
|
|
|
|
for _, h := range matched {
|
|
h(event)
|
|
}
|
|
}
|
|
|
|
// PublishAsync dispatches an event to all matching subscribers
|
|
// in separate goroutines. Useful for I/O-heavy handlers.
|
|
func (b *Bus) PublishAsync(event Event) {
|
|
b.mu.RLock()
|
|
var matched []Handler
|
|
for pattern, subs := range b.subs {
|
|
if match(event.Label, pattern) {
|
|
for _, s := range subs {
|
|
matched = append(matched, s.handler)
|
|
}
|
|
}
|
|
}
|
|
b.mu.RUnlock()
|
|
|
|
for _, h := range matched {
|
|
go h(event)
|
|
}
|
|
}
|
|
|
|
// match checks if a concrete label matches a subscription pattern.
|
|
//
|
|
// "chat.message.abc" matches "chat.message.abc" (exact)
|
|
// "chat.message.abc" matches "chat.message.*" (wildcard)
|
|
// "chat.message.abc" matches "chat.*" (wildcard)
|
|
// "chat.message.abc" matches "*" (global wildcard)
|
|
func match(label, pattern string) bool {
|
|
if pattern == "*" {
|
|
return true
|
|
}
|
|
if pattern == label {
|
|
return true
|
|
}
|
|
if !strings.HasSuffix(pattern, "*") {
|
|
return false
|
|
}
|
|
prefix := pattern[:len(pattern)-1] // "chat.message." from "chat.message.*"
|
|
return strings.HasPrefix(label, prefix)
|
|
}
|