fix: pubsub leak and shutdown seq

This commit is contained in:
adamdottv 2025-05-05 14:23:29 -05:00
parent afcdabd095
commit 3cc08494a5
No known key found for this signature in database
GPG key ID: 9CB48779AF150E75
3 changed files with 170 additions and 16 deletions

View file

@ -64,22 +64,27 @@ func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] {
b.subs[sub] = struct{}{}
b.subCount++
go func() {
<-ctx.Done()
// Only start a goroutine if the context can actually be canceled
if ctx.Done() != nil {
go func() {
<-ctx.Done()
b.mu.Lock()
defer b.mu.Unlock()
b.mu.Lock()
defer b.mu.Unlock()
select {
case <-b.done:
return
default:
}
select {
case <-b.done:
return
default:
}
delete(b.subs, sub)
close(sub)
b.subCount--
}()
if _, exists := b.subs[sub]; exists {
delete(b.subs, sub)
close(sub)
b.subCount--
}
}()
}
return sub
}