package events import ( "context" "sort" "strconv" "sync" "sync/atomic" "time" ) var globalEventSeq atomic.Uint64 // Bus publishes runtime events and creates filtered channels. type Bus interface { Publish(ctx context.Context, evt Event) PublishResult Channel() EventChannel Close() error Stats() Stats } // PublishResult reports per-publish delivery outcomes. type PublishResult struct { Matched int Delivered int Dropped int Blocked int Closed bool } // EventBus is an in-process runtime event broadcaster. type EventBus struct { mu sync.RWMutex subs map[uint64]*eventSubscription closed bool nextSubID atomic.Uint64 published atomic.Uint64 matched atomic.Uint64 delivered atomic.Uint64 dropped atomic.Uint64 blocked atomic.Uint64 } var _ Bus = (*EventBus)(nil) // NewBus creates an in-process runtime event bus. func NewBus() *EventBus { return &EventBus{ subs: make(map[uint64]*eventSubscription), } } // Publish broadcasts evt to subscriptions whose filters match it. func (b *EventBus) Publish(ctx context.Context, evt Event) PublishResult { if b == nil { return PublishResult{Closed: true} } if ctx == nil { ctx = context.Background() } if evt.Time.IsZero() { evt.Time = time.Now() } if evt.ID == "" { evt.ID = nextEventID() } subs, closed := b.snapshotSubscribers() if closed { return PublishResult{Closed: true} } b.published.Add(1) result := PublishResult{} for _, sub := range subs { if !matchesFilters(sub.filters, evt) { continue } result.Matched++ b.matched.Add(1) delivery := sub.enqueue(ctx, evt) if delivery.closed { continue } result.Delivered += delivery.delivered result.Dropped += delivery.dropped result.Blocked += delivery.blocked b.delivered.Add(uint64(delivery.delivered)) b.dropped.Add(uint64(delivery.dropped)) b.blocked.Add(uint64(delivery.blocked)) } return result } // Channel returns the root event channel for this bus. func (b *EventBus) Channel() EventChannel { return eventChannel{bus: b} } // Close closes the bus and all active subscriptions. func (b *EventBus) Close() error { if b == nil { return nil } b.mu.Lock() if b.closed { b.mu.Unlock() return nil } b.closed = true subs := make([]*eventSubscription, 0, len(b.subs)) for id, sub := range b.subs { subs = append(subs, sub) delete(b.subs, id) } b.mu.Unlock() for _, sub := range subs { sub.closeInput() } return nil } // Stats returns a snapshot of bus and subscription counters. func (b *EventBus) Stats() Stats { if b == nil { return Stats{Closed: true} } b.mu.RLock() closed := b.closed subs := make([]*eventSubscription, 0, len(b.subs)) for _, sub := range b.subs { subs = append(subs, sub) } b.mu.RUnlock() sortSubscriptions(subs) stats := Stats{ Published: b.published.Load(), Matched: b.matched.Load(), Delivered: b.delivered.Load(), Dropped: b.dropped.Load(), Blocked: b.blocked.Load(), Closed: closed, Subscribers: len(subs), SubscriberStats: make([]SubscriberStats, 0, len(subs)), } for _, sub := range subs { stats.SubscriberStats = append(stats.SubscriberStats, sub.Stats()) } return stats } func (b *EventBus) subscribe( ctx context.Context, filters []Filter, opts SubscribeOptions, handler Handler, once bool, ) (Subscription, error) { if b == nil { return nil, ErrBusClosed } id := b.nextSubID.Add(1) sub := newSubscription(b, id, filters, opts, handler, once) b.mu.Lock() if b.closed { b.mu.Unlock() sub.closeInput() return nil, ErrBusClosed } b.subs[id] = sub b.mu.Unlock() if handler != nil { go sub.run(ctx) } sub.watchContext(ctx) return sub, nil } func (b *EventBus) unsubscribe(id uint64) { b.mu.Lock() sub, ok := b.subs[id] if ok { delete(b.subs, id) } b.mu.Unlock() if ok { sub.closeInput() } } func (b *EventBus) snapshotSubscribers() ([]*eventSubscription, bool) { b.mu.RLock() defer b.mu.RUnlock() if b.closed { return nil, true } subs := make([]*eventSubscription, 0, len(b.subs)) for _, sub := range b.subs { subs = append(subs, sub) } sortSubscriptions(subs) return subs, false } func sortSubscriptions(subs []*eventSubscription) { sort.Slice(subs, func(i, j int) bool { if subs[i].opts.Priority == subs[j].opts.Priority { return subs[i].id < subs[j].id } return subs[i].opts.Priority > subs[j].opts.Priority }) } func nextEventID() string { id := globalEventSeq.Add(1) return "evt-" + strconv.FormatUint(id, 10) }