git.iko.soy/crush
on main help sign in
fix(pubsub): respect channelBufferSize parameter in Subsribe
abeaff0Yeonuk Hwang
internal/pubsub/broker.go+11−9
compare with main
unifiedsplit
@@ -8,11 +8,12 @@ import (
8const bufferSize = 64
8const bufferSize = 64
9
9
10type Broker[T any] struct {
10type Broker[T any] struct {
11− subs map[chan Event[T]]struct{}
11+ subs map[chan Event[T]]struct{}
12− mu sync.RWMutex
12+ mu sync.RWMutex
13− done chan struct{}
13+ done chan struct{}
14− subCount int
14+ subCount int
15− maxEvents int
15+ maxEvents int
16+ channelBufferSize int
16}
17}
17
18
18func NewBroker[T any]() *Broker[T] {
19func NewBroker[T any]() *Broker[T] {
@@ -21,9 +22,10 @@ func NewBroker[T any]() *Broker[T] {
21
22
22func NewBrokerWithOptions[T any](channelBufferSize, maxEvents int) *Broker[T] {
23func NewBrokerWithOptions[T any](channelBufferSize, maxEvents int) *Broker[T] {
23 return &Broker[T]{
24 return &Broker[T]{
24− subs: make(map[chan Event[T]]struct{}),
25+ subs: make(map[chan Event[T]]struct{}),
25− done: make(chan struct{}),
26+ done: make(chan struct{}),
26− maxEvents: maxEvents,
27+ maxEvents: maxEvents,
28+ channelBufferSize: channelBufferSize,
27 }
29 }
28}
30}
29
31
@@ -58,7 +60,7 @@ func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] {
58 default:
60 default:
59 }
61 }
60
62
61− sub := make(chan Event[T], bufferSize)
63+ sub := make(chan Event[T], b.channelBufferSize)
62 b.subs[sub] = struct{}{}
64 b.subs[sub] = struct{}{}
63 b.subCount++
65 b.subCount++
64
66