refactor(eventbus): improve channel and pipe management logic

- Replace atomic counter with simple increment for CowMap length calculation
- Change channel stopCh type from chan any to chan struct{} for proper signaling
- Store reflect.Value of topic in channel struct to avoid repeated conversions
- Simplify transfer method by removing redundant topic parameter
- Use LoadOrStore to prevent concurrent channel creation and properly close duplicates
- Modify Pipe handlers map to use handler pointer as key via reflect.ValueOf().Pointer()
- Replace sending struct{} on stopCh with channel close for graceful shutdown
- Correct PublishSync to call synchronous publish method in singleton wrapper

Author: werbenhu

.claude/settings.local.json

@@ -0,0 +1,7 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(go test ./... -count=1)"
+    ]
+  }
+}

cowmap.go

@@ -2,7 +2,6 @@ package eventbus
 
 import (
 	"sync"
-	"sync/atomic"
 )
 
 // CowMap is a wrapper of Copy-On-Write map
@@ -26,7 +25,7 @@ func NewCowMap() *CowMap {
 func (c *CowMap) Len() uint32 {
 	var size uint32
 	c.Range(func(k, v any) bool {
-		atomic.AddUint32(&size, 1)
+		size++
 		return true
 	})
 	return size

eventbus.go

@@ -10,10 +10,11 @@ type channel struct {
 	sync.RWMutex
 	bufferSize int
 	topic      string
+	topicValue reflect.Value
 	channel    chan any
 	handlers   *CowMap
 	closed     bool
-	stopCh     chan any
+	stopCh     chan struct{}
 }
 
 // newChannel creates a new channel with a specified topic and buffer size.
@@ -28,20 +29,20 @@ func newChannel(topic string, bufferSize int) *channel {
 	}
 	c := &channel{
 		topic:      topic,
+		topicValue: reflect.ValueOf(topic),
 		bufferSize: bufferSize,
 		channel:    ch,
 		handlers:   NewCowMap(),
-		stopCh:     make(chan any),
+		stopCh:     make(chan struct{}),
 	}
 	go c.loop()
 	return c
 }
 
 // transfer calls all the handlers in the channel with the given payload.
 // It iterates over the handlers in the handlers map to call them with the payload.
-func (c *channel) transfer(topic string, payload any) {
+func (c *channel) transfer(payload any) {
 	var payloadValue reflect.Value
-	topicValue := reflect.ValueOf(c.topic)
 
 	c.handlers.Range(func(key any, fn any) bool {
 		handler := fn.(*reflect.Value)
@@ -55,7 +56,7 @@ func (c *channel) transfer(topic string, payload any) {
 		} else {
 			payloadValue = reflect.ValueOf(payload)
 		}
-		(*handler).Call([]reflect.Value{topicValue, payloadValue})
+		(*handler).Call([]reflect.Value{c.topicValue, payloadValue})
 		return true
 	})
 }
@@ -67,7 +68,7 @@ func (c *channel) loop() {
 	for {
 		select {
 		case payload := <-c.channel:
-			c.transfer(c.topic, payload)
+			c.transfer(payload)
 		case <-c.stopCh:
 			return
 		}
@@ -95,7 +96,7 @@ func (c *channel) publishSync(payload any) error {
 	if c.closed {
 		return ErrChannelClosed
 	}
-	c.transfer(c.topic, payload)
+	c.transfer(payload)
 	return nil
 }
 
@@ -132,7 +133,7 @@ func (c *channel) close() {
 		return
 	}
 	c.closed = true
-	c.stopCh <- struct{}{}
+	close(c.stopCh)
 	c.handlers.Clear()
 	close(c.channel)
 }
@@ -193,8 +194,12 @@ func (e *EventBus) Subscribe(topic string, handler any) error {
 
 	ch, ok := e.channels.Load(topic)
 	if !ok {
-		ch = newChannel(topic, e.bufferSize)
-		e.channels.Store(topic, ch)
+		newCh := newChannel(topic, e.bufferSize)
+		ch, loaded := e.channels.LoadOrStore(topic, newCh)
+		if loaded {
+			newCh.close()
+		}
+		return ch.(*channel).subscribe(handler)
 	}
 	return ch.(*channel).subscribe(handler)
 }
@@ -205,10 +210,13 @@ func (e *EventBus) Subscribe(topic string, handler any) error {
 // The type of the payload must correspond to the second parameter of the handler in `Subscribe()`.
 func (e *EventBus) Publish(topic string, payload any) error {
 	ch, ok := e.channels.Load(topic)
-
 	if !ok {
-		ch = newChannel(topic, e.bufferSize)
-		e.channels.Store(topic, ch)
+		newCh := newChannel(topic, e.bufferSize)
+		var loaded bool
+		ch, loaded = e.channels.LoadOrStore(topic, newCh)
+		if loaded {
+			newCh.close()
+		}
 	}
 
 	return ch.(*channel).publish(payload)
@@ -219,10 +227,13 @@ func (e *EventBus) Publish(topic string, payload any) error {
 // It does not use channels and instead directly calls the handler function.
 func (e *EventBus) PublishSync(topic string, payload any) error {
 	ch, ok := e.channels.Load(topic)
-
 	if !ok {
-		ch = newChannel(topic, e.bufferSize)
-		e.channels.Store(topic, ch)
+		newCh := newChannel(topic, e.bufferSize)
+		var loaded bool
+		ch, loaded = e.channels.LoadOrStore(topic, newCh)
+		if loaded {
+			newCh.close()
+		}
 	}
 
 	return ch.(*channel).publishSync(payload)

pipe.go

@@ -1,6 +1,7 @@
 package eventbus
 
 import (
+	"reflect"
 	"sync"
 )
 
@@ -16,15 +17,15 @@ type Pipe[T any] struct {
 	channel    chan T
 	handlers   *CowMap
 	closed     bool
-	stopCh     chan any
+	stopCh     chan struct{}
 }
 
 // NewPipe create a unbuffered pipe
 func NewPipe[T any]() *Pipe[T] {
 	p := &Pipe[T]{
 		bufferSize: -1,
 		channel:    make(chan T),
-		stopCh:     make(chan any),
+		stopCh:     make(chan struct{}),
 		handlers:   NewCowMap(),
 	}
 
@@ -42,7 +43,7 @@ func NewBufferedPipe[T any](bufferSize int) *Pipe[T] {
 	p := &Pipe[T]{
 		bufferSize: bufferSize,
 		channel:    make(chan T, bufferSize),
-		stopCh:     make(chan any),
+		stopCh:     make(chan struct{}),
 		handlers:   NewCowMap(),
 	}
 
@@ -72,7 +73,8 @@ func (p *Pipe[T]) Subscribe(handler Handler[T]) error {
 	if p.closed {
 		return ErrChannelClosed
 	}
-	p.handlers.Store(&handler, handler)
+	key := reflect.ValueOf(handler).Pointer()
+	p.handlers.Store(key, handler)
 	return nil
 }
 
@@ -83,7 +85,8 @@ func (p *Pipe[T]) Unsubscribe(handler Handler[T]) error {
 	if p.closed {
 		return ErrChannelClosed
 	}
-	p.handlers.Delete(&handler)
+	key := reflect.ValueOf(handler).Pointer()
+	p.handlers.Delete(key)
 	return nil
 }
 
@@ -122,6 +125,6 @@ func (p *Pipe[T]) Close() {
 		return
 	}
 	p.closed = true
-	p.stopCh <- struct{}{}
+	close(p.stopCh)
 	close(p.channel)
 }

singleton.go

@@ -40,7 +40,7 @@ func Publish(topic string, payload any) error {
 // PublishSync is a synchronous version of Publish that triggers the handlers defined for a topic with the given payload.
 // The type of the payload must correspond to the second parameter of the handler in `Subscribe()`.
 func PublishSync(topic string, payload any) error {
-	return singleton.Publish(topic, payload)
+	return singleton.PublishSync(topic, payload)
 }
 
 // Close closes the singleton instance of EventBus.