feat(eventbus): enhance event handling with error propagation and global bus management

internal/pkg/eventbus/bus.go

@@ -2,13 +2,15 @@ package eventbus
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"log/slog"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
 )
 
-type Handler[T any] func(ctx context.Context, event T)
+type Handler[T any] func(ctx context.Context, event T) error
 
 type ErrorHandler func(ctx context.Context, err error)
 
@@ -51,6 +53,10 @@ func WithAsyncConcurrency[T any](n int) Option[T] {
 
 // Subscribe 返回 handlerID 和取消订阅函数
 func (b *Bus[T]) Subscribe(h Handler[T]) (uint64, func()) {
+	if h == nil {
+		panic("eventbus: nil handler")
+	}
+
 	id := atomic.AddUint64(&b.nextID, 1)
 
 	b.mu.Lock()
@@ -63,11 +69,19 @@ func (b *Bus[T]) Subscribe(h Handler[T]) (uint64, func()) {
 }
 
 func (b *Bus[T]) SubscribeOnce(h Handler[T]) (uint64, func()) {
-	var id uint64
+	if h == nil {
+		panic("eventbus: nil handler")
+	}
 
-	wrapper := func(ctx context.Context, event T) {
+	var id uint64
+	var called atomic.Bool
+
+	wrapper := func(ctx context.Context, event T) error {
+		if !called.CompareAndSwap(false, true) {
+			return nil
+		}
 		b.Unsubscribe(id)
-		h(ctx, event)
+		return h(ctx, event)
 	}
 
 	id = atomic.AddUint64(&b.nextID, 1)
@@ -87,26 +101,21 @@ func (b *Bus[T]) Unsubscribe(id uint64) {
 	b.mu.Unlock()
 }
 
-func (b *Bus[T]) Publish(ctx context.Context, event T) {
+func (b *Bus[T]) Publish(ctx context.Context, event T) error {
 	handlers := b.snapshotHandlers()
+	errs := make([]error, 0, len(handlers))
 	for _, h := range handlers {
-		b.callHandler(ctx, h, event)
+		if err := b.callHandler(ctx, h, event); err != nil {
+			errs = append(errs, err)
+		}
 	}
+	return errors.Join(errs...)
 }
 
 func (b *Bus[T]) PublishAsync(ctx context.Context, event T) {
 	handlers := b.snapshotHandlers()
 	for _, h := range handlers {
-		if b.asyncSem != nil {
+		go b.callHandlerAsync(ctx, h, event)
-			b.asyncSem <- struct{}{}
-			go func() {
-				defer func() { <-b.asyncSem }()
-				b.callHandler(ctx, h, event)
-			}()
-			continue
-		}
-
-		go b.callHandler(ctx, h, event)
 	}
 }
 
@@ -127,13 +136,41 @@ func (b *Bus[T]) snapshotHandlers() []Handler[T] {
 	return handlers
 }
 
-func (b *Bus[T]) callHandler(ctx context.Context, h Handler[T], event T) {
+func (b *Bus[T]) callHandlerAsync(ctx context.Context, h Handler[T], event T) {
+	if b.asyncSem != nil {
+		select {
+		case b.asyncSem <- struct{}{}:
+			defer func() { <-b.asyncSem }()
+		case <-ctx.Done():
+			b.handleError(ctx, ctx.Err())
+			return
+		}
+	}
+
+	if err := b.callHandler(ctx, h, event); err != nil {
+		b.handleError(ctx, err)
+	}
+}
+
+func (b *Bus[T]) callHandler(ctx context.Context, h Handler[T], event T) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			if b.onError != nil {
+			err = fmt.Errorf("event handler panic: %v\n%s", r, debug.Stack())
-				b.onError(ctx, fmt.Errorf("event handler panic: %v\n%s", r, debug.Stack()))
-			}
 		}
 	}()
-	h(ctx, event)
+	if err = h(ctx, event); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (b *Bus[T]) handleError(ctx context.Context, err error) {
+	if err == nil {
+		return
+	}
+	if b.onError != nil {
+		b.onError(ctx, err)
+		return
+	}
+	slog.Error("eventbus handler failed", "error", err)
 }

internal/pkg/eventbus/bus_test.go

@@ -0,0 +1,144 @@
+package eventbus
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type testEvent struct {
+	ID int64
+}
+
+func TestPackageSubscribeAndPublish(t *testing.T) {
+	resetManagerForTest(t)
+
+	var got atomic.Int64
+	_, unsubscribe := Subscribe(func(ctx context.Context, event testEvent) error {
+		got.Store(event.ID)
+		return nil
+	})
+	defer unsubscribe()
+
+	if err := Publish(context.Background(), testEvent{ID: 42}); err != nil {
+		t.Fatalf("publish failed: %v", err)
+	}
+
+	if got.Load() != 42 {
+		t.Fatalf("expected event ID 42, got %d", got.Load())
+	}
+}
+
+func TestPublishReturnsJoinedHandlerErrors(t *testing.T) {
+	firstErr := errors.New("first")
+	secondErr := errors.New("second")
+	bus := New[testEvent]()
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		return firstErr
+	})
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		return secondErr
+	})
+
+	err := bus.Publish(context.Background(), testEvent{})
+	if !errors.Is(err, firstErr) {
+		t.Fatalf("expected joined error to include first error, got %v", err)
+	}
+	if !errors.Is(err, secondErr) {
+		t.Fatalf("expected joined error to include second error, got %v", err)
+	}
+}
+
+func TestPanicReturnsError(t *testing.T) {
+	bus := New[testEvent]()
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		panic("boom")
+	})
+
+	err := bus.Publish(context.Background(), testEvent{})
+	if err == nil {
+		t.Fatalf("expected panic to be returned as error")
+	}
+}
+
+func TestPublishAsyncCallsErrorHandler(t *testing.T) {
+	handlerErr := errors.New("async failed")
+	var handled atomic.Int64
+	done := make(chan struct{})
+	bus := New[testEvent](WithErrorHandler[testEvent](func(ctx context.Context, err error) {
+		if errors.Is(err, handlerErr) {
+			handled.Add(1)
+		}
+		close(done)
+	}))
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		return handlerErr
+	})
+
+	bus.PublishAsync(context.Background(), testEvent{})
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("expected async error handler to be called")
+	}
+
+	if handled.Load() != 1 {
+		t.Fatalf("expected error handler to be called once, got %d", handled.Load())
+	}
+}
+
+func TestSubscribeOnceConcurrentPublishOnlyCallsOnce(t *testing.T) {
+	bus := New[testEvent]()
+	var calls atomic.Int64
+
+	bus.SubscribeOnce(func(ctx context.Context, event testEvent) error {
+		calls.Add(1)
+		time.Sleep(10 * time.Millisecond)
+		return nil
+	})
+
+	const workers = 32
+	var wg sync.WaitGroup
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = bus.Publish(context.Background(), testEvent{})
+		}()
+	}
+	wg.Wait()
+
+	if calls.Load() != 1 {
+		t.Fatalf("expected once handler to be called once, got %d", calls.Load())
+	}
+}
+
+func TestPublishAsyncAllowsReentrantPublishWithConcurrencyLimit(t *testing.T) {
+	bus := New[testEvent](WithAsyncConcurrency[testEvent](1))
+	done := make(chan struct{})
+	var calls atomic.Int64
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		if calls.Add(1) == 1 {
+			bus.PublishAsync(ctx, testEvent{ID: 2})
+			return nil
+		}
+		close(done)
+		return nil
+	})
+
+	bus.PublishAsync(context.Background(), testEvent{ID: 1})
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("expected reentrant async publish to complete")
+	}
+}

internal/pkg/eventbus/manager.go

@@ -1,6 +1,7 @@
 package eventbus
 
 import (
+	"context"
 	"reflect"
 	"sync"
 )
@@ -10,6 +11,24 @@ var (
 	buss = make(map[reflect.Type]any)
 )
 
+// Register initializes the global bus for T. If the bus already exists, the
+// existing instance is returned and later options are ignored.
+func Register[T any](opts ...Option[T]) *Bus[T] {
+	key := eventTypeOf[T]()
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	if bus, ok := buss[key]; ok {
+		return bus.(*Bus[T])
+	}
+
+	created := New[T](opts...)
+	buss[key] = created
+	return created
+}
+
+// Get returns the global bus for T, creating it with default options if needed.
 func Get[T any]() *Bus[T] {
 	key := eventTypeOf[T]()
 
@@ -39,3 +58,23 @@ func eventTypeOf[T any]() reflect.Type {
 	}
 	return typ
 }
+
+// Subscribe registers a handler on the global bus for T.
+func Subscribe[T any](h Handler[T]) (uint64, func()) {
+	return Get[T]().Subscribe(h)
+}
+
+// SubscribeOnce registers a handler that runs at most once on the global bus for T.
+func SubscribeOnce[T any](h Handler[T]) (uint64, func()) {
+	return Get[T]().SubscribeOnce(h)
+}
+
+// Publish synchronously publishes event to the global bus for T.
+func Publish[T any](ctx context.Context, event T) error {
+	return Get[T]().Publish(ctx, event)
+}
+
+// PublishAsync asynchronously publishes event to the global bus for T.
+func PublishAsync[T any](ctx context.Context, event T) {
+	Get[T]().PublishAsync(ctx, event)
+}

internal/pkg/eventbus/manager_test.go

@@ -1,6 +1,7 @@
 package eventbus
 
 import (
+	"context"
 	"reflect"
 	"sync"
 	"testing"
@@ -59,6 +60,31 @@ func TestGetCreatesOnlyOneBusUnderConcurrency(t *testing.T) {
 	}
 }
 
+func TestRegisterReturnsConfiguredGlobalBus(t *testing.T) {
+	resetManagerForTest(t)
+
+	var handled bool
+	registered := Register[testEvent](WithErrorHandler[testEvent](func(ctx context.Context, err error) {
+		handled = true
+	}))
+	got := Get[testEvent]()
+
+	if registered != got {
+		t.Fatalf("expected registered bus to be returned by Get")
+	}
+
+	got.handleError(context.Background(), assertErr{})
+	if !handled {
+		t.Fatalf("expected registered error handler to be used")
+	}
+}
+
+type assertErr struct{}
+
+func (assertErr) Error() string {
+	return "assert"
+}
+
 func resetManagerForTest(t *testing.T) {
 	t.Helper()