server: fix event dropping - use blocking callback instead of TrySend
The pubsub library's SubscribeChan uses non-blocking TrySend which drops events when the channel is full. Switch to Subscribe with a blocking callback to ensure no events are lost. Also increased channel buffers from 64 to 256.
This commit is contained in:
parent
cd94cdf259
commit
79ab165ffd
|
|
@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue