From df13d5ced5b772658427bbb9a4deb6952233e569 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Tue, 21 Apr 2026 11:50:43 +0800 Subject: [PATCH] feat(eventbus): enhance event handling with error propagation and global bus management --- internal/pkg/eventbus/bus.go | 79 ++++++++++---- internal/pkg/eventbus/bus_test.go | 144 ++++++++++++++++++++++++++ internal/pkg/eventbus/manager.go | 39 +++++++ internal/pkg/eventbus/manager_test.go | 26 +++++ 4 files changed, 267 insertions(+), 21 deletions(-) create mode 100644 internal/pkg/eventbus/bus_test.go diff --git a/internal/pkg/eventbus/bus.go b/internal/pkg/eventbus/bus.go index 80f096e..f576c7f 100644 --- a/internal/pkg/eventbus/bus.go +++ b/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 { - b.asyncSem <- struct{}{} - go func() { - defer func() { <-b.asyncSem }() - b.callHandler(ctx, h, event) - }() - continue - } - - go b.callHandler(ctx, h, event) + go b.callHandlerAsync(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 { - b.onError(ctx, fmt.Errorf("event handler panic: %v\n%s", r, debug.Stack())) - } + err = 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) } diff --git a/internal/pkg/eventbus/bus_test.go b/internal/pkg/eventbus/bus_test.go new file mode 100644 index 0000000..6590f00 --- /dev/null +++ b/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") + } +} diff --git a/internal/pkg/eventbus/manager.go b/internal/pkg/eventbus/manager.go index 7514799..23caf7f 100644 --- a/internal/pkg/eventbus/manager.go +++ b/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) +} diff --git a/internal/pkg/eventbus/manager_test.go b/internal/pkg/eventbus/manager_test.go index 958afe2..3b24e75 100644 --- a/internal/pkg/eventbus/manager_test.go +++ b/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()