diff --git a/cmd/olliesrv/internal/session/event.go b/cmd/olliesrv/internal/session/event.go index 4474d01..5f91d26 100644 --- a/cmd/olliesrv/internal/session/event.go +++ b/cmd/olliesrv/internal/session/event.go @@ -31,9 +31,16 @@ func PublishEvent(topic, payload string) { } // SubscribeEvents returns a channel that receives all events matching the topic. +// Events are never dropped — publishing blocks if the channel is full. func SubscribeEvents(ctx context.Context, topic string) <-chan Event { - ch := make(chan Event, 64) - sub := eventBus.SubscribeChan(topic, ch, pubsub.KeepOpen) + // Use a callback instead of SubscribeChan to avoid non-blocking TrySend + ch := make(chan Event, 256) + sub := eventBus.Subscribe(topic, func(ev Event) { + select { + case ch <- ev: + case <-ctx.Done(): + } + }) go func() { <-ctx.Done() eventBus.Unsubscribe(sub) @@ -48,7 +55,7 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event { func SubscribeEventsFiltered(ctx context.Context, filter string) <-chan Event { // Subscribe to all events and filter client-side for complex patterns // (pubsub only supports simple topic matching) - ch := make(chan Event, 64) + ch := make(chan Event, 256) allEvents := SubscribeEvents(ctx, "*") go func() { defer close(ch) @@ -56,7 +63,8 @@ func SubscribeEventsFiltered(ctx context.Context, filter string) <-chan Event { if MatchTopic(filter, ev.Topic) { select { case ch <- ev: - default: // drop if full + case <-ctx.Done(): + return } } }