From fd24e3191267a47924dc128f7e8a5ca57fcf5445 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 23 Feb 2026 12:18:13 +0800 Subject: [PATCH] Update Makefile for enhanced testing and add new event documentation - Modify unit test commands in the Makefile to include additional skip patterns for memory leak tests, improving test coverage and accuracy. - Expand benchmark and memory leak detection to include the event module, ensuring comprehensive testing across all components. - Add new design and TODO documentation files for the event module to facilitate future development. --- .gitignore | 2 + Makefile | 16 +- event/README.md | 184 ++++++++++++++++++ event/bench_test.go | 382 +++++++++++++++++++++++++++++++++++++ event/bus.go | 159 ++++++++++++++++ event/bus_test.go | 320 +++++++++++++++++++++++++++++++ event/leak_test.go | 348 ++++++++++++++++++++++++++++++++++ event/listener.go | 141 ++++++++++++++ event/listener_test.go | 222 ++++++++++++++++++++++ event/option.go | 51 +++++ event/queue.go | 208 ++++++++++++++++++++ event/queue_test.go | 387 ++++++++++++++++++++++++++++++++++++++ event/service.go | 221 ++++++++++++++++++++++ event/service_test.go | 229 ++++++++++++++++++++++ event/sub.go | 101 ++++++++++ event/sub_test.go | 181 ++++++++++++++++++ event/types/interfaces.go | 23 +++ event/types/types.go | 102 ++++++++++ event/types/types_test.go | 267 ++++++++++++++++++++++++++ event/worker.go | 94 +++++++++ event/worker_test.go | 213 +++++++++++++++++++++ 21 files changed, 3843 insertions(+), 8 deletions(-) create mode 100644 event/README.md create mode 100644 event/bench_test.go create mode 100644 event/bus.go create mode 100644 event/bus_test.go create mode 100644 event/leak_test.go create mode 100644 event/listener.go create mode 100644 event/listener_test.go create mode 100644 event/option.go create mode 100644 event/queue.go create mode 100644 event/queue_test.go create mode 100644 event/service.go create mode 100644 event/service_test.go create mode 100644 event/sub.go create mode 100644 event/sub_test.go create mode 100644 event/types/interfaces.go create mode 100644 event/types/types.go create mode 100644 event/types/types_test.go create mode 100644 event/worker.go create mode 100644 event/worker_test.go diff --git a/.gitignore b/.gitignore index d67186a4..e899992a 100644 --- a/.gitignore +++ b/.gitignore @@ -66,3 +66,5 @@ release/* sandbox/TODO-VNC.md sandbox/docker/chrome/PLAN.md sandbox/DESIGN-REMOTE.md +event/DESIGN.md +event/TODO.md diff --git a/Makefile b/Makefile index b852bd93..cae35a18 100644 --- a/Makefile +++ b/Makefile @@ -30,7 +30,7 @@ TESTTAGS ?= "" unit-test: echo "mode: count" > coverage.out for d in $(TESTFOLDER); do \ - $(GO) test -tags $(TESTTAGS) -v -covermode=count -coverprofile=profile.out -coverpkg=$$(echo $$d | sed "s/\/test$$//g") -skip='TestMemoryLeak|TestIsolateDisposal' $$d > tmp.out; \ + $(GO) test -tags $(TESTTAGS) -v -covermode=count -coverprofile=profile.out -coverpkg=$$(echo $$d | sed "s/\/test$$//g") -skip='TestMemoryLeak|TestIsolateDisposal|TestLeak_|TestScenario_' $$d > tmp.out; \ cat tmp.out; \ if grep -q "^--- FAIL" tmp.out; then \ rm tmp.out; \ @@ -56,7 +56,7 @@ unit-test: unit-test-core: echo "mode: count" > coverage.out for d in $(TESTFOLDER_CORE); do \ - $(GO) test -tags $(TESTTAGS) -v -covermode=count -coverprofile=profile.out -coverpkg=$$(echo $$d | sed "s/\/test$$//g") -skip='TestMemoryLeak|TestIsolateDisposal' $$d > tmp.out; \ + $(GO) test -tags $(TESTTAGS) -v -covermode=count -coverprofile=profile.out -coverpkg=$$(echo $$d | sed "s/\/test$$//g") -skip='TestMemoryLeak|TestIsolateDisposal|TestLeak_|TestScenario_' $$d > tmp.out; \ cat tmp.out; \ if grep -q "^--- FAIL" tmp.out; then \ rm tmp.out; \ @@ -224,9 +224,9 @@ unit-test-sandbox: benchmark: @echo "" @echo "=============================================" - @echo "Running Benchmark Tests (agent & trace)..." + @echo "Running Benchmark Tests (agent, trace, event)..." @echo "=============================================" - @for d in $$($(GO) list ./agent/... ./trace/...); do \ + @for d in $$($(GO) list ./agent/... ./trace/... ./event/...); do \ if $(GO) test -list=Benchmark $$d 2>/dev/null | grep -q "^Benchmark"; then \ echo ""; \ echo "πŸ“Š Benchmarking: $$d"; \ @@ -244,14 +244,14 @@ benchmark: memory-leak: @echo "" @echo "=============================================" - @echo "Running Memory Leak Detection (agent & trace)..." + @echo "Running Memory Leak Detection (agent, trace, event)..." @echo "=============================================" - @for d in $$($(GO) list ./agent/... ./trace/...); do \ - if $(GO) test -list='TestMemoryLeak|TestIsolateDisposal|TestGoroutineLeak' $$d 2>/dev/null | grep -qE "^Test(MemoryLeak|IsolateDisposal|GoroutineLeak)"; then \ + @for d in $$($(GO) list ./agent/... ./trace/... ./event/...); do \ + if $(GO) test -list='TestMemoryLeak|TestIsolateDisposal|TestGoroutineLeak|TestLeak_|TestScenario_' $$d 2>/dev/null | grep -qE "^Test(MemoryLeak|IsolateDisposal|GoroutineLeak|Leak_|Scenario_)"; then \ echo ""; \ echo "πŸ” Memory Leak Detection: $$d"; \ echo "---------------------------------------------"; \ - $(GO) test -run='TestMemoryLeak|TestIsolateDisposal|TestGoroutineLeak' -v -timeout=5m $$d || exit 1; \ + $(GO) test -run='TestMemoryLeak|TestIsolateDisposal|TestGoroutineLeak|TestLeak_|TestScenario_' -v -timeout=5m $$d || exit 1; \ fi; \ done @echo "" diff --git a/event/README.md b/event/README.md new file mode 100644 index 00000000..36a4f660 --- /dev/null +++ b/event/README.md @@ -0,0 +1,184 @@ +# event β€” Yao In-Process Event Bus + +Global event service for async/sync event routing, serial queue processing, and real-time subscriptions. All operations are goroutine-safe. + +## Import + +```go +import ( + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) +``` + +## Core Concepts + +| Concept | Description | +|---|---| +| **Push** | Async fire-and-forget delivery. Returns event ID immediately. | +| **Call** | Sync request-response. Blocks until handler writes to `resp`. | +| **Handler** | One per prefix (e.g. `"trace"`). Processes `Push` and `Call` events. | +| **Queue** | FIFO serial processing per entity (e.g. per traceID). Events in same queue never run concurrently. | +| **Listener** | Persistent background consumer (registered at startup). Gets a copy of every matching event. | +| **Subscriber** | Dynamic subscription (e.g. SSE/WebSocket). Non-blocking; skips if channel full. | + +## Lifecycle + +```go +// 1. Register handlers and listeners (before Start, typically in init()) +event.Register("trace", traceHandler, event.MaxWorkers(512), event.ReservedWorkers(20)) +event.Register("job", jobHandler) +event.Listen("trace.*", traceListener) + +// 2. Start +event.Start() + +// 3. Use (from any goroutine) +event.Push(ctx, "trace.add", payload, event.Queue(traceQueueID)) +id, data, err := event.Call(ctx, "trace.get", req, event.Queue(traceQueueID)) + +// 4. Stop (during shutdown) +event.Stop(ctx) +``` + +## Handler + +Implement `types.Handler`: + +```go +type TraceHandler struct{} + +func (h *TraceHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + var p TracePayload + if err := ev.Should(&p); err != nil { + if ev.IsCall { resp <- types.Result{Err: err} } + return + } + // ... business logic ... + if ev.IsCall { + resp <- types.Result{Data: result} + } +} + +func (h *TraceHandler) Shutdown(ctx context.Context) error { return nil } +``` + +- `ctx`: non-cancellable for Push; caller's context for Call. +- `resp`: always non-nil. Write exactly once for Call; ignore for Push. +- `ev.Should(&target)`: type-safe payload extraction. +- Panics are recovered automatically; `ErrHandlerPanic` is returned to Call. + +## Queue + +```go +queueID, err := event.QueueCreate("trace") // auto-generated ID +queueID, err := event.QueueCreate("trace", "my-id") // custom ID + +event.Push(ctx, "trace.add", data, event.Queue(queueID)) // serial +event.Call(ctx, "trace.get", req, event.Queue(queueID)) // serial, same queue + +event.QueueRelease(queueID) // graceful: drain pending, reject new +event.QueueAbort(queueID) // forceful: discard pending, reject new +``` + +## Listener + +Implement `types.Listener`: + +```go +type MailListener struct{} +func (l *MailListener) OnEvent(ev *types.Event) { /* ... */ } +func (l *MailListener) Shutdown(ctx context.Context) error { return nil } + +// Register before Start +event.Listen("mail.*", &MailListener{}, event.Filter(fn), event.BufferSize(4096)) +``` + +- Each listener runs in its own goroutine. +- Non-blocking: if buffer full, event is skipped (logged as warning). + +## Subscriber + +```go +ch := make(chan *types.Event, 256) +subID := event.Subscribe("trace.*", ch, event.Filter(fn)) +defer event.Unsubscribe(subID) + +for ev := range ch { + // push to SSE / WebSocket +} +``` + +- Non-blocking: if `ch` full, event is skipped silently. +- Call `Unsubscribe` when client disconnects. + +## Context Propagation + +```go +ctx = event.WithSID(ctx, sessionID) +ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: "u-1"}) + +// Inside handler: +sid := ev.SID +auth := ev.Auth // may be nil +``` + +SID and Auth are extracted from `ctx` automatically when calling `Push`/`Call`. + +## Pattern Matching + +Used by `Listen` and `Subscribe`: + +| Pattern | Matches | +|---|---| +| `"*"` | Everything | +| `"trace.*"` | `"trace.add"`, `"trace.get"`, etc. | +| `"trace.add"` | Exact match only | + +## Handler Options + +| Option | Default | Description | +|---|---|---| +| `MaxWorkers(n)` | 512 | Max concurrent goroutines for this handler | +| `ReservedWorkers(n)` | 10 | Slots reserved for Call (Push can use Maxβˆ’Reserved) | +| `QueueSize(n)` | 8192 | Per-queue buffered channel capacity | + +## Errors + +| Error | When | +|---|---| +| `ErrNotStarted` | Push/Call before Start or after Stop | +| `ErrNoHandler` | No handler registered for event prefix | +| `ErrQueueFull` | Queue buffer at capacity | +| `ErrQueueNotFound` | Queue ID never created | +| `ErrQueueReleased` | Queue already released/aborted | +| `ErrQueueExists` | QueueCreate with duplicate ID | +| `ErrHandlerPanic` | Handler panicked (recovered) | + +## Performance (M2 Max, 12 cores) + +| Metric | Value | +|---|---| +| Push (no queue) | ~860K ops/sec, 456 B/op | +| Call (no queue) | ~1.2M ops/sec, 440 B/op | +| Push (with queue) | ~2.9M ops/sec, 341 B/op | +| 1000-user scenario (2000 queues, 27K events) | ~100K events/sec, 280ms total | +| Steady-state memory (1000 users) | ~27 MB | +| Goroutine leaks | Zero | + +## File Structure + +``` +event/ +β”œβ”€β”€ types/ +β”‚ β”œβ”€β”€ types.go # Event, Result, HandlerEntry, FilterEntry, options +β”‚ └── interfaces.go # Handler, Listener interfaces +β”œβ”€β”€ service.go # Register, Start, Stop, Reload, global state +β”œβ”€β”€ bus.go # Push, Call, QueueCreate/Release/Abort +β”œβ”€β”€ queue.go # FIFO queue + queue manager +β”œβ”€β”€ worker.go # Worker pool (two-tier semaphore) +β”œβ”€β”€ listener.go # Listener manager + pattern matching +β”œβ”€β”€ sub.go # Subscriber manager +β”œβ”€β”€ option.go # Option functions +└── README.md +``` diff --git a/event/bench_test.go b/event/bench_test.go new file mode 100644 index 00000000..4538af32 --- /dev/null +++ b/event/bench_test.go @@ -0,0 +1,382 @@ +package event_test + +import ( + "context" + "fmt" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --------------------------------------------------------------------------- +// Shared bench handler: lightweight, simulates minimal real work. +// --------------------------------------------------------------------------- + +type benchHandler struct { + processed atomic.Int64 +} + +func (h *benchHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + h.processed.Add(1) + if ev.IsCall { + resp <- types.Result{Data: "ok"} + } +} + +func (h *benchHandler) Shutdown(ctx context.Context) error { return nil } + +// benchListener counts received events. +type benchListener struct { + received atomic.Int64 +} + +func (l *benchListener) OnEvent(ev *types.Event) { l.received.Add(1) } +func (l *benchListener) Shutdown(ctx context.Context) error { return nil } + +// --------------------------------------------------------------------------- +// Benchmark: Push throughput (no queue, pure worker dispatch) +// --------------------------------------------------------------------------- + +func BenchmarkPush_NoQueue(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("bench", h, event.MaxWorkers(512)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + ctx := context.Background() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + _, _ = event.Push(ctx, "bench.work", nil) + } + }) + b.StopTimer() + + // Drain workers + time.Sleep(100 * time.Millisecond) + b.ReportMetric(float64(h.processed.Load()), "events_handled") +} + +// --------------------------------------------------------------------------- +// Benchmark: Call throughput (no queue, synchronous round-trip) +// --------------------------------------------------------------------------- + +func BenchmarkCall_NoQueue(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("bench", h, event.MaxWorkers(512)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + ctx := context.Background() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + _, _, _ = event.Call(ctx, "bench.get", nil) + } + }) +} + +// --------------------------------------------------------------------------- +// Benchmark: Push throughput with Queue (serial per queue) +// --------------------------------------------------------------------------- + +func BenchmarkPush_WithQueue(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("bench", h, event.MaxWorkers(512), event.QueueSize(8192)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + qID, _ := event.QueueCreate("bench") + b.Cleanup(func() { event.QueueRelease(qID) }) + + ctx := context.Background() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = event.Push(ctx, "bench.work", nil, event.Queue(qID)) + } + b.StopTimer() + + event.QueueRelease(qID) + time.Sleep(200 * time.Millisecond) +} + +// --------------------------------------------------------------------------- +// Scenario: 1000 concurrent users, each with trace + job queues. +// +// Simulates: +// - 1000 users Γ— 2 queues (trace + job) = 2000 queues +// - Each user pushes 20 trace events + 5 job events + 1 Call per queue +// - 200 SSE subscribers watching "trace.*" and "job.*" +// - 2 Listeners (trace.* + job.*) +// +// Reports: total duration, events/sec, memory delta. +// --------------------------------------------------------------------------- + +func TestScenario_1000Users(t *testing.T) { + event.Reset() + defer event.Reset() + + traceH := &benchHandler{} + jobH := &benchHandler{} + event.Register("trace", traceH, event.MaxWorkers(512), event.ReservedWorkers(20), event.QueueSize(8192)) + event.Register("job", jobH, event.MaxWorkers(256), event.ReservedWorkers(10), event.QueueSize(4096)) + + traceL := &benchListener{} + jobL := &benchListener{} + event.Listen("trace.*", traceL) + event.Listen("job.*", jobL) + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + const ( + numUsers = 1000 + tracePushPerUser = 20 + jobPushPerUser = 5 + callsPerQueue = 1 + numSubscribers = 200 + subscriberBufSize = 256 + ) + + // --- Subscribers --- + subChans := make([]chan *types.Event, numSubscribers) + subIDs := make([]string, numSubscribers) + for i := 0; i < numSubscribers; i++ { + ch := make(chan *types.Event, subscriberBufSize) + subChans[i] = ch + pattern := "trace.*" + if i%2 == 1 { + pattern = "job.*" + } + subIDs[i] = event.Subscribe(pattern, ch) + } + defer func() { + for _, id := range subIDs { + event.Unsubscribe(id) + } + }() + + // Drain subscribers in background + var subReceived atomic.Int64 + subDone := make(chan struct{}) + go func() { + defer close(subDone) + for _, ch := range subChans { + go func(c chan *types.Event) { + for range c { + subReceived.Add(1) + } + }(ch) + } + }() + + // --- Memory before --- + runtime.GC() + var memBefore runtime.MemStats + runtime.ReadMemStats(&memBefore) + + // --- Run --- + start := time.Now() + var wg sync.WaitGroup + + for u := 0; u < numUsers; u++ { + wg.Add(1) + go func(userID int) { + defer wg.Done() + ctx := event.WithSID(context.Background(), fmt.Sprintf("sess-%d", userID)) + ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: fmt.Sprintf("u-%d", userID)}) + + // Create trace queue + traceQID, err := event.QueueCreate("trace") + if err != nil { + t.Errorf("user %d: trace QueueCreate: %v", userID, err) + return + } + + // Create job queue + jobQID, err := event.QueueCreate("job") + if err != nil { + t.Errorf("user %d: job QueueCreate: %v", userID, err) + event.QueueRelease(traceQID) + return + } + + // Push trace events + for i := 0; i < tracePushPerUser; i++ { + _, _ = event.Push(ctx, "trace.add", i, event.Queue(traceQID)) + } + + // Push job events + for i := 0; i < jobPushPerUser; i++ { + _, _ = event.Push(ctx, "job.progress", i, event.Queue(jobQID)) + } + + // Call on each queue + for i := 0; i < callsPerQueue; i++ { + callCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + _, _, _ = event.Call(callCtx, "trace.get", nil, event.Queue(traceQID)) + cancel() + + callCtx2, cancel2 := context.WithTimeout(ctx, 5*time.Second) + _, _, _ = event.Call(callCtx2, "job.status", nil, event.Queue(jobQID)) + cancel2() + } + + // Release queues + event.QueueRelease(traceQID) + event.QueueRelease(jobQID) + }(u) + } + + wg.Wait() + elapsed := time.Since(start) + + // Wait for queues to drain + time.Sleep(500 * time.Millisecond) + + // --- Memory after --- + runtime.GC() + var memAfter runtime.MemStats + runtime.ReadMemStats(&memAfter) + + // --- Results --- + totalPush := int64(numUsers) * int64(tracePushPerUser+jobPushPerUser) + totalCall := int64(numUsers) * int64(callsPerQueue) * 2 + totalEvents := totalPush + totalCall + traceProcessed := traceH.processed.Load() + jobProcessed := jobH.processed.Load() + listenerTrace := traceL.received.Load() + listenerJob := jobL.received.Load() + memDeltaMB := float64(memAfter.TotalAlloc-memBefore.TotalAlloc) / 1024 / 1024 + + t.Logf("=== 1000-User Scenario Results ===") + t.Logf("Users: %d", numUsers) + t.Logf("Queues created: %d (trace: %d, job: %d)", numUsers*2, numUsers, numUsers) + t.Logf("Subscribers: %d", numSubscribers) + t.Logf("Total events: %d (push: %d, call: %d)", totalEvents, totalPush, totalCall) + t.Logf("Trace processed: %d", traceProcessed) + t.Logf("Job processed: %d", jobProcessed) + t.Logf("Listener trace: %d", listenerTrace) + t.Logf("Listener job: %d", listenerJob) + t.Logf("Sub received: %d", subReceived.Load()) + t.Logf("Elapsed: %v", elapsed) + t.Logf("Throughput: %.0f events/sec", float64(totalEvents)/elapsed.Seconds()) + t.Logf("Memory delta: %.2f MB (TotalAlloc)", memDeltaMB) + + // --- Assertions --- + expectedProcessed := totalPush + totalCall + actualProcessed := traceProcessed + jobProcessed + if actualProcessed < expectedProcessed { + t.Errorf("processed %d < expected %d (some events lost)", actualProcessed, expectedProcessed) + } + + if elapsed > 30*time.Second { + t.Errorf("scenario took %v, expected < 30s", elapsed) + } +} + +// --------------------------------------------------------------------------- +// Benchmark: Queue create/release churn (lifecycle overhead) +// --------------------------------------------------------------------------- + +func BenchmarkQueueCreateRelease(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("bench", h, event.QueueSize(64)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + qID, err := event.QueueCreate("bench") + if err != nil { + b.Fatalf("QueueCreate: %v", err) + } + event.QueueRelease(qID) + } + }) +} + +// --------------------------------------------------------------------------- +// Benchmark: Subscriber notify throughput (fanout to 200 subscribers) +// --------------------------------------------------------------------------- + +func BenchmarkSubscriberFanout(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("bench", h, event.MaxWorkers(512)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + const numSubs = 200 + for i := 0; i < numSubs; i++ { + ch := make(chan *types.Event, 1024) + event.Subscribe("bench.*", ch) + go func(c chan *types.Event) { + for range c { + } + }(ch) + } + + ctx := context.Background() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + _, _ = event.Push(ctx, "bench.work", nil) + } + }) +} + +// --------------------------------------------------------------------------- +// Benchmark: Mixed Push/Call with 2000 queues (1000 users Γ— 2) +// --------------------------------------------------------------------------- + +func BenchmarkMixed_2000Queues(b *testing.B) { + event.Reset() + h := &benchHandler{} + event.Register("mix", h, event.MaxWorkers(512), event.ReservedWorkers(20), event.QueueSize(4096)) + _ = event.Start() + b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() }) + + const numQueues = 2000 + queueIDs := make([]string, numQueues) + for i := 0; i < numQueues; i++ { + qID, err := event.QueueCreate("mix") + if err != nil { + b.Fatalf("QueueCreate %d: %v", i, err) + } + queueIDs[i] = qID + } + b.Cleanup(func() { + for _, qID := range queueIDs { + event.QueueRelease(qID) + } + time.Sleep(200 * time.Millisecond) + }) + + ctx := context.Background() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + i := 0 + for pb.Next() { + qID := queueIDs[i%numQueues] + if i%10 == 0 { + callCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + _, _, _ = event.Call(callCtx, "mix.get", nil, event.Queue(qID)) + cancel() + } else { + _, _ = event.Push(ctx, "mix.work", nil, event.Queue(qID)) + } + i++ + } + }) +} diff --git a/event/bus.go b/event/bus.go new file mode 100644 index 00000000..34517c3b --- /dev/null +++ b/event/bus.go @@ -0,0 +1,159 @@ +package event + +import ( + "context" + "fmt" + "strings" + "sync/atomic" + + "github.com/yaoapp/yao/event/types" +) + +var eventIDCounter atomic.Uint64 + +func nextEventID() string { + id := eventIDCounter.Add(1) + return fmt.Sprintf("ev-%d", id) +} + +// prefixOf extracts the handler prefix from an event type. +// "trace.add" -> "trace", "job.progress" -> "job" +func prefixOf(typ string) string { + if i := strings.IndexByte(typ, '.'); i >= 0 { + return typ[:i] + } + return typ +} + +// Push delivers an event asynchronously (fire-and-forget). +// SID and Auth are extracted from ctx automatically. +// Returns the auto-generated event ID. +func Push(ctx context.Context, typ string, payload any, opts ...types.PushOption) (string, error) { + prefix := prefixOf(typ) + entry, pool, err := getHandler(prefix) + if err != nil { + return "", err + } + _ = entry // used for queue config lookup + + ev := &types.Event{ + Type: typ, + ID: nextEventID(), + IsCall: false, + Payload: payload, + SID: SIDFrom(ctx), + Auth: AuthFrom(ctx), + } + for _, opt := range opts { + opt(ev) + } + + // Notify listeners and subscribers (non-blocking, before handler) + svc.lmgr.notify(ev) + svc.smgr.notify(ev) + + // Route to queue or direct dispatch + if ev.Queue != "" { + q, err := svc.queues.get(ev.Queue) + if err != nil { + return ev.ID, err + } + discard := make(chan types.Result, 1) + if err := q.enqueue(ctx, ev, discard); err != nil { + return ev.ID, err + } + return ev.ID, nil + } + + // No queue: direct dispatch with discard channel + discard := make(chan types.Result, 1) + pushCtx := context.WithoutCancel(ctx) + if _, err := pool.dispatch(pushCtx, ev, discard); err != nil { + return ev.ID, fmt.Errorf("event push: worker unavailable: %w", err) + } + return ev.ID, nil +} + +// Call delivers an event synchronously and blocks until the handler responds. +// SID and Auth are extracted from ctx automatically. +// Returns the auto-generated event ID and the handler's result. +func Call(ctx context.Context, typ string, payload any, opts ...types.PushOption) (string, any, error) { + prefix := prefixOf(typ) + _, pool, err := getHandler(prefix) + if err != nil { + return "", nil, err + } + + ev := &types.Event{ + Type: typ, + ID: nextEventID(), + IsCall: true, + Payload: payload, + SID: SIDFrom(ctx), + Auth: AuthFrom(ctx), + } + for _, opt := range opts { + opt(ev) + } + + // Notify listeners and subscribers + svc.lmgr.notify(ev) + svc.smgr.notify(ev) + + resp := make(chan types.Result, 1) + + if ev.Queue != "" { + q, err := svc.queues.get(ev.Queue) + if err != nil { + return ev.ID, nil, err + } + if err := q.enqueue(ctx, ev, resp); err != nil { + return ev.ID, nil, err + } + } else { + if _, err := pool.dispatch(ctx, ev, resp); err != nil { + return ev.ID, nil, fmt.Errorf("event call: worker unavailable: %w", err) + } + } + + // Wait for handler result or context cancellation + select { + case result := <-resp: + return ev.ID, result.Data, result.Err + case <-ctx.Done(): + return ev.ID, nil, ctx.Err() + } +} + +// QueueCreate creates a new event queue bound to a handler prefix. +// Returns the queue ID. If no id is provided, one is auto-generated. +func QueueCreate(prefix string, id ...string) (string, error) { + entry, pool, err := getHandler(prefix) + if err != nil { + return "", err + } + + queueID := "" + if len(id) > 0 && id[0] != "" { + queueID = id[0] + } else { + queueID = fmt.Sprintf("q-%s-%d", prefix, eventIDCounter.Add(1)) + } + + if err := svc.queues.create(prefix, queueID, entry.QueueSize, pool); err != nil { + return "", err + } + return queueID, nil +} + +// QueueRelease gracefully releases a queue (async). +// Rejects new events immediately; existing events are drained internally. +func QueueRelease(queueID string) { + svc.queues.release(queueID) +} + +// QueueAbort forcefully releases a queue (async). +// Rejects new events, discards pending events, waits for in-flight to finish. +func QueueAbort(queueID string) { + svc.queues.abortOne(queueID) +} diff --git a/event/bus_test.go b/event/bus_test.go new file mode 100644 index 00000000..97de5f05 --- /dev/null +++ b/event/bus_test.go @@ -0,0 +1,320 @@ +package event_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --- Test handler --- + +type recordHandler struct { + mu sync.Mutex + calls []string // records ev.Type for each Handle call + shutdown bool +} + +func (h *recordHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + h.mu.Lock() + h.calls = append(h.calls, ev.Type) + h.mu.Unlock() + + if ev.IsCall { + var p string + if err := ev.Should(&p); err == nil { + resp <- types.Result{Data: "echo:" + p} + } else { + resp <- types.Result{Data: "echo:" + ev.Type} + } + } +} + +func (h *recordHandler) Shutdown(ctx context.Context) error { + h.mu.Lock() + defer h.mu.Unlock() + h.shutdown = true + return nil +} + +func (h *recordHandler) getCalls() []string { + h.mu.Lock() + defer h.mu.Unlock() + cp := make([]string, len(h.calls)) + copy(cp, h.calls) + return cp +} + +// --- Phase 3: Push / Call basic routing (no queue) --- + +func TestPush_NoQueue(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &recordHandler{} + event.Register("foo", h) + if err := event.Start(); err != nil { + t.Fatal(err) + } + defer func() { _ = event.Stop(context.Background()) }() + + id, err := event.Push(context.Background(), "foo.bar", "payload1") + if err != nil { + t.Fatalf("Push failed: %v", err) + } + if id == "" { + t.Fatal("expected non-empty event ID") + } + + // Wait for async handler + time.Sleep(50 * time.Millisecond) + calls := h.getCalls() + if len(calls) != 1 || calls[0] != "foo.bar" { + t.Fatalf("expected [foo.bar], got %v", calls) + } +} + +func TestCall_NoQueue(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &recordHandler{} + event.Register("foo", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + id, data, err := event.Call(context.Background(), "foo.get", "hello") + if err != nil { + t.Fatalf("Call failed: %v", err) + } + if id == "" { + t.Fatal("expected non-empty event ID") + } + if data != "echo:hello" { + t.Fatalf("expected echo:hello, got %v", data) + } +} + +func TestPush_UnregisteredPrefix(t *testing.T) { + event.Reset() + defer event.Reset() + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, err := event.Push(context.Background(), "unknown.thing", nil) + if err != event.ErrNoHandler { + t.Fatalf("expected ErrNoHandler, got %v", err) + } +} + +func TestPush_NotStarted(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _, err := event.Push(context.Background(), "foo.bar", nil) + if err != event.ErrNotStarted { + t.Fatalf("expected ErrNotStarted, got %v", err) + } +} + +func TestPush_SIDAndAuth(t *testing.T) { + event.Reset() + defer event.Reset() + + var captured *types.Event + var mu sync.Mutex + + h := &captureHandler{onHandle: func(ev *types.Event) { + mu.Lock() + captured = ev + mu.Unlock() + }} + event.Register("foo", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ctx := event.WithSID(context.Background(), "sess-abc") + ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: "u-1"}) + + _, err := event.Push(ctx, "foo.bar", "data") + if err != nil { + t.Fatalf("Push failed: %v", err) + } + + time.Sleep(50 * time.Millisecond) + mu.Lock() + defer mu.Unlock() + if captured == nil { + t.Fatal("handler was not called") + } + if captured.SID != "sess-abc" { + t.Fatalf("expected SID sess-abc, got %s", captured.SID) + } + if captured.Auth == nil || captured.Auth.UserID != "u-1" { + t.Fatalf("expected Auth.UserID u-1, got %+v", captured.Auth) + } +} + +// captureHandler captures the event for inspection. +type captureHandler struct { + onHandle func(*types.Event) +} + +func (h *captureHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + if h.onHandle != nil { + h.onHandle(ev) + } + if ev.IsCall { + resp <- types.Result{Data: "ok"} + } +} + +func (h *captureHandler) Shutdown(ctx context.Context) error { return nil } + +// --- Coverage: prefixOf without dot --- + +func TestPush_TypeWithoutDot(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &recordHandler{} + event.Register("nodot", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + id, err := event.Push(context.Background(), "nodot", "payload") + if err != nil { + t.Fatalf("Push failed: %v", err) + } + if id == "" { + t.Fatal("expected non-empty event ID") + } + time.Sleep(50 * time.Millisecond) + calls := h.getCalls() + if len(calls) != 1 || calls[0] != "nodot" { + t.Fatalf("expected [nodot], got %v", calls) + } +} + +// --- Coverage: Call unregistered prefix --- + +func TestCall_UnregisteredPrefix(t *testing.T) { + event.Reset() + defer event.Reset() + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _, err := event.Call(context.Background(), "unknown.thing", nil) + if err != event.ErrNoHandler { + t.Fatalf("expected ErrNoHandler, got %v", err) + } +} + +// --- Coverage: Call with queue (happy path) --- + +func TestCall_WithQueue(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &recordHandler{} + event.Register("foo", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, err := event.QueueCreate("foo") + if err != nil { + t.Fatalf("QueueCreate failed: %v", err) + } + defer event.QueueRelease(qID) + + id, data, err := event.Call(context.Background(), "foo.get", "hello", event.Queue(qID)) + if err != nil { + t.Fatalf("Call with queue failed: %v", err) + } + if id == "" { + t.Fatal("expected non-empty event ID") + } + if data != "echo:hello" { + t.Fatalf("expected echo:hello, got %v", data) + } +} + +// --- Coverage: Call with non-existent queue --- + +func TestCall_QueueNotFound(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _, err := event.Call(context.Background(), "foo.get", nil, event.Queue("no-such-queue")) + if err != event.ErrQueueNotFound { + t.Fatalf("expected ErrQueueNotFound, got %v", err) + } +} + +// --- Coverage: Call ctx timeout --- + +func TestCall_CtxTimeout(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &captureHandler{onHandle: func(ev *types.Event) { + time.Sleep(500 * time.Millisecond) + }} + event.Register("slow", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + _, _, err := event.Call(ctx, "slow.op", nil) + if err == nil { + t.Fatal("expected timeout error") + } +} + +// --- Coverage: Call no-queue dispatch failure (ctx cancelled) --- + +func TestCall_NoQueue_DispatchFail(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &concurrencyHandler{ + peak: &atomic.Int32{}, + current: &atomic.Int32{}, + delay: 200 * time.Millisecond, + } + event.Register("tiny", h, event.MaxWorkers(1), event.ReservedWorkers(0)) + _ = event.Start() + + // Saturate the single total slot with a Call in background + bgDone := make(chan struct{}) + go func() { + defer close(bgDone) + _, _, _ = event.Call(context.Background(), "tiny.work", nil) + }() + time.Sleep(10 * time.Millisecond) + + // Another Call with already-cancelled context should fail at dispatch + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _, err := event.Call(ctx, "tiny.op", nil) + if err == nil { + t.Fatal("expected error for cancelled ctx call") + } + + // Wait for background goroutine to finish before Stop + <-bgDone + _ = event.Stop(context.Background()) +} diff --git a/event/leak_test.go b/event/leak_test.go new file mode 100644 index 00000000..244419d3 --- /dev/null +++ b/event/leak_test.go @@ -0,0 +1,348 @@ +package event_test + +import ( + "context" + "fmt" + "runtime" + "sync" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --------------------------------------------------------------------------- +// Helper: snapshot goroutine count after GC stabilization. +// --------------------------------------------------------------------------- + +func stableGoroutineCount() int { + // Let runtime settle: GC + finalizers + scheduler + for i := 0; i < 5; i++ { + runtime.GC() + runtime.Gosched() + time.Sleep(10 * time.Millisecond) + } + return runtime.NumGoroutine() +} + +// leakHandler is a no-op handler for leak tests. +type leakHandler struct{} + +func (h *leakHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + if ev.IsCall { + resp <- types.Result{Data: "ok"} + } +} + +func (h *leakHandler) Shutdown(ctx context.Context) error { return nil } + +// leakListener is a no-op listener for leak tests. +type leakListener struct{} + +func (l *leakListener) OnEvent(ev *types.Event) {} +func (l *leakListener) Shutdown(ctx context.Context) error { return nil } + +// --------------------------------------------------------------------------- +// Test: 1000 Queue create/release cycles leak no goroutines. +// --------------------------------------------------------------------------- + +func TestLeak_QueueCreateRelease(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("leak", &leakHandler{}, event.QueueSize(64)) + _ = event.Start() + + before := stableGoroutineCount() + + const cycles = 1000 + for i := 0; i < cycles; i++ { + qID, err := event.QueueCreate("leak") + if err != nil { + t.Fatalf("cycle %d: QueueCreate: %v", i, err) + } + // Push a few events to exercise consumer goroutine + for j := 0; j < 3; j++ { + _, _ = event.Push(context.Background(), "leak.work", j, event.Queue(qID)) + } + event.QueueRelease(qID) + } + + // Let all consumer goroutines drain and exit + time.Sleep(500 * time.Millisecond) + after := stableGoroutineCount() + + _ = event.Stop(context.Background()) + + leaked := after - before + t.Logf("goroutines: before=%d after=%d delta=%d (over %d cycles)", before, after, leaked, cycles) + + // Allow a small margin for runtime jitter (GC, timers, etc.) + if leaked > 5 { + t.Errorf("goroutine leak: %d goroutines accumulated over %d queue cycles", leaked, cycles) + } +} + +// --------------------------------------------------------------------------- +// Test: 1000 Queue create/abort cycles leak no goroutines. +// --------------------------------------------------------------------------- + +func TestLeak_QueueCreateAbort(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("leak", &leakHandler{}, event.QueueSize(64)) + _ = event.Start() + + before := stableGoroutineCount() + + const cycles = 1000 + for i := 0; i < cycles; i++ { + qID, err := event.QueueCreate("leak") + if err != nil { + t.Fatalf("cycle %d: QueueCreate: %v", i, err) + } + for j := 0; j < 3; j++ { + _, _ = event.Push(context.Background(), "leak.work", j, event.Queue(qID)) + } + event.QueueAbort(qID) + } + + time.Sleep(500 * time.Millisecond) + after := stableGoroutineCount() + + _ = event.Stop(context.Background()) + + leaked := after - before + t.Logf("goroutines: before=%d after=%d delta=%d (over %d cycles)", before, after, leaked, cycles) + + if leaked > 5 { + t.Errorf("goroutine leak: %d goroutines accumulated over %d abort cycles", leaked, cycles) + } +} + +// --------------------------------------------------------------------------- +// Test: Subscriber create/unsubscribe cycles leak no goroutines or memory. +// --------------------------------------------------------------------------- + +func TestLeak_SubscriberLifecycle(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("leak", &leakHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + before := stableGoroutineCount() + runtime.GC() + var memBefore runtime.MemStats + runtime.ReadMemStats(&memBefore) + + const cycles = 1000 + for i := 0; i < cycles; i++ { + ch := make(chan *types.Event, 16) + subID := event.Subscribe("leak.*", ch) + + _, _ = event.Push(context.Background(), "leak.work", nil) + time.Sleep(time.Microsecond) // let notify propagate + + event.Unsubscribe(subID) + } + + time.Sleep(200 * time.Millisecond) + after := stableGoroutineCount() + + runtime.GC() + var memAfter runtime.MemStats + runtime.ReadMemStats(&memAfter) + + leaked := after - before + memDeltaMB := float64(int64(memAfter.HeapInuse)-int64(memBefore.HeapInuse)) / 1024 / 1024 + + t.Logf("goroutines: before=%d after=%d delta=%d", before, after, leaked) + t.Logf("heap in-use delta: %.2f MB", memDeltaMB) + + if leaked > 3 { + t.Errorf("goroutine leak: %d goroutines after %d sub/unsub cycles", leaked, cycles) + } +} + +// --------------------------------------------------------------------------- +// Test: Start/Stop cycles leak no goroutines. +// --------------------------------------------------------------------------- + +func TestLeak_StartStopCycles(t *testing.T) { + before := stableGoroutineCount() + + const cycles = 20 + for i := 0; i < cycles; i++ { + event.Reset() + event.Register("leak", &leakHandler{}) + event.Listen("leak.*", &leakListener{}) + _ = event.Start() + + ctx := context.Background() + for j := 0; j < 10; j++ { + _, _ = event.Push(ctx, "leak.work", j) + } + time.Sleep(10 * time.Millisecond) + + _ = event.Stop(ctx) + } + event.Reset() + + time.Sleep(300 * time.Millisecond) + after := stableGoroutineCount() + + leaked := after - before + t.Logf("goroutines: before=%d after=%d delta=%d (over %d start/stop cycles)", before, after, leaked, cycles) + + if leaked > 3 { + t.Errorf("goroutine leak: %d goroutines after %d start/stop cycles", leaked, cycles) + } +} + +// --------------------------------------------------------------------------- +// Test: 1000 concurrent users creating/using/releasing queues, verify +// no goroutine leak when everything settles. +// --------------------------------------------------------------------------- + +func TestLeak_1000Users_FullCycle(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("trace", &leakHandler{}, event.MaxWorkers(512), event.QueueSize(8192)) + event.Register("job", &leakHandler{}, event.MaxWorkers(256), event.QueueSize(4096)) + event.Listen("trace.*", &leakListener{}) + _ = event.Start() + + before := stableGoroutineCount() + + const numUsers = 1000 + var wg sync.WaitGroup + for u := 0; u < numUsers; u++ { + wg.Add(1) + go func(uid int) { + defer wg.Done() + ctx := event.WithSID(context.Background(), fmt.Sprintf("s-%d", uid)) + + tqID, err := event.QueueCreate("trace") + if err != nil { + return + } + jqID, err := event.QueueCreate("job") + if err != nil { + event.QueueRelease(tqID) + return + } + + for i := 0; i < 10; i++ { + _, _ = event.Push(ctx, "trace.add", i, event.Queue(tqID)) + } + for i := 0; i < 3; i++ { + _, _ = event.Push(ctx, "job.progress", i, event.Queue(jqID)) + } + + callCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + _, _, _ = event.Call(callCtx, "trace.get", nil, event.Queue(tqID)) + cancel() + + event.QueueRelease(tqID) + event.QueueRelease(jqID) + }(u) + } + + wg.Wait() + time.Sleep(1 * time.Second) // let all consumers drain + + after := stableGoroutineCount() + + _ = event.Stop(context.Background()) + + // Final check after full stop + afterStop := stableGoroutineCount() + + leaked := after - before + leakedAfterStop := afterStop - before + + t.Logf("goroutines: before=%d after_drain=%d after_stop=%d", before, after, afterStop) + t.Logf("delta after drain: %d, delta after stop: %d", leaked, leakedAfterStop) + + if leaked > 10 { + t.Errorf("goroutine leak after drain: %d (1000 users Γ— 2 queues)", leaked) + } + if leakedAfterStop > 3 { + t.Errorf("goroutine leak after stop: %d", leakedAfterStop) + } +} + +// --------------------------------------------------------------------------- +// Test: Memory stability under sustained load. +// Push 100k events through 100 queues, measure heap growth. +// --------------------------------------------------------------------------- + +func TestLeak_MemoryStability(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("mem", &leakHandler{}, event.MaxWorkers(256), event.QueueSize(8192)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + const ( + numQueues = 100 + eventsPerQueue = 1000 + totalEvents = numQueues * eventsPerQueue + ) + + queueIDs := make([]string, numQueues) + for i := 0; i < numQueues; i++ { + qID, _ := event.QueueCreate("mem") + queueIDs[i] = qID + } + + runtime.GC() + var memBefore runtime.MemStats + runtime.ReadMemStats(&memBefore) + + ctx := context.Background() + var wg sync.WaitGroup + for q := 0; q < numQueues; q++ { + wg.Add(1) + go func(qIdx int) { + defer wg.Done() + qID := queueIDs[qIdx] + for i := 0; i < eventsPerQueue; i++ { + _, _ = event.Push(ctx, "mem.work", i, event.Queue(qID)) + } + }(q) + } + wg.Wait() + + // Release all and wait + for _, qID := range queueIDs { + event.QueueRelease(qID) + } + time.Sleep(1 * time.Second) + + runtime.GC() + var memAfter runtime.MemStats + runtime.ReadMemStats(&memAfter) + + // Use signed arithmetic to handle GC reclaiming memory between snapshots. + heapDeltaMB := float64(int64(memAfter.HeapInuse)-int64(memBefore.HeapInuse)) / 1024 / 1024 + allocDeltaMB := float64(memAfter.TotalAlloc-memBefore.TotalAlloc) / 1024 / 1024 + + t.Logf("=== Memory Stability ===") + t.Logf("Events: %d (%d queues Γ— %d events)", totalEvents, numQueues, eventsPerQueue) + t.Logf("HeapInuse delta: %.2f MB", heapDeltaMB) + t.Logf("TotalAlloc: %.2f MB", allocDeltaMB) + t.Logf("Alloc/event: %.0f bytes", allocDeltaMB*1024*1024/float64(totalEvents)) + + // After drain, heap should not retain significant memory. + // Allow generous 50 MB for 100k events (runtime overhead, GC timing). + if heapDeltaMB > 50 { + t.Errorf("heap grew %.2f MB after %d events, possible leak", heapDeltaMB, totalEvents) + } +} diff --git a/event/listener.go b/event/listener.go new file mode 100644 index 00000000..371088c7 --- /dev/null +++ b/event/listener.go @@ -0,0 +1,141 @@ +package event + +import ( + "context" + "strings" + "sync" + + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/event/types" +) + +// listenerEntry holds a registered listener with its filter configuration. +type listenerEntry struct { + pattern string + listener types.Listener + filter func(*types.Event) bool + bufferSize int + ch chan *types.Event + done chan struct{} +} + +// listenerManager manages all registered listeners. +type listenerManager struct { + mu sync.RWMutex + entries []*listenerEntry + started bool +} + +func newListenerManager() *listenerManager { + return &listenerManager{} +} + +// register adds a listener. Must be called before start(). +func (lm *listenerManager) register(pattern string, listener types.Listener, opts ...types.FilterOption) { + fe := &types.FilterEntry{ + Pattern: pattern, + BufferSize: types.DefaultBufferSize, + } + for _, opt := range opts { + opt(fe) + } + + lm.mu.Lock() + defer lm.mu.Unlock() + lm.entries = append(lm.entries, &listenerEntry{ + pattern: pattern, + listener: listener, + filter: fe.Filter, + bufferSize: fe.BufferSize, + }) +} + +// start creates channels and goroutines for each listener. +func (lm *listenerManager) start() { + lm.mu.Lock() + defer lm.mu.Unlock() + + for _, entry := range lm.entries { + entry.ch = make(chan *types.Event, entry.bufferSize) + entry.done = make(chan struct{}) + go lm.consume(entry) + } + lm.started = true +} + +// consume is the goroutine that reads from a listener's channel. +func (lm *listenerManager) consume(entry *listenerEntry) { + defer close(entry.done) + for ev := range entry.ch { + func() { + defer func() { + if r := recover(); r != nil { + log.Error("event listener panic: pattern=%s type=%s err=%v", entry.pattern, ev.Type, r) + } + }() + entry.listener.OnEvent(ev) + }() + } +} + +// notify sends an event to all matching listeners (non-blocking). +func (lm *listenerManager) notify(ev *types.Event) { + lm.mu.RLock() + defer lm.mu.RUnlock() + + if !lm.started { + return + } + + for _, entry := range lm.entries { + if !matchPattern(entry.pattern, ev.Type) { + continue + } + if entry.filter != nil && !entry.filter(ev) { + continue + } + select { + case entry.ch <- ev: + default: + log.Warn("event listener buffer full: pattern=%s type=%s id=%s (skipped)", entry.pattern, ev.Type, ev.ID) + } + } +} + +// stop shuts down all listeners. +func (lm *listenerManager) stop(ctx context.Context) { + lm.mu.Lock() + lm.started = false + entries := lm.entries + lm.mu.Unlock() + + for _, entry := range entries { + close(entry.ch) + } + for _, entry := range entries { + <-entry.done + _ = entry.listener.Shutdown(ctx) + } +} + +// matchPattern matches an event type against a listener/subscriber pattern. +// - "*" matches everything +// - "foo.*" matches any type starting with "foo." +// - "foo.bar" matches exactly "foo.bar" +func matchPattern(pattern, eventType string) bool { + if pattern == "*" { + return true + } + if strings.HasSuffix(pattern, ".*") { + prefix := strings.TrimSuffix(pattern, "*") + return strings.HasPrefix(eventType, prefix) + } + return pattern == eventType +} + +// Listen registers a persistent listener. Must be called before Start. +func Listen(pattern string, listener types.Listener, opts ...types.FilterOption) { + svc.mu.Lock() + defer svc.mu.Unlock() + svc.lmgr.register(pattern, listener, opts...) +} diff --git a/event/listener_test.go b/event/listener_test.go new file mode 100644 index 00000000..37bbee0e --- /dev/null +++ b/event/listener_test.go @@ -0,0 +1,222 @@ +package event_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --- Phase 6: Listener tests --- + +// collectListener collects received events. +type collectListener struct { + mu sync.Mutex + events []*types.Event + shut bool +} + +func (l *collectListener) OnEvent(ev *types.Event) { + l.mu.Lock() + defer l.mu.Unlock() + l.events = append(l.events, ev) +} + +func (l *collectListener) Shutdown(ctx context.Context) error { + l.mu.Lock() + defer l.mu.Unlock() + l.shut = true + return nil +} + +func (l *collectListener) getEvents() []*types.Event { + l.mu.Lock() + defer l.mu.Unlock() + cp := make([]*types.Event, len(l.events)) + copy(cp, l.events) + return cp +} + +func TestListener_PatternMatch(t *testing.T) { + event.Reset() + defer event.Reset() + + allL := &collectListener{} + fooL := &collectListener{} + exactL := &collectListener{} + + event.Listen("*", allL) + event.Listen("foo.*", fooL) + event.Listen("foo.exact", exactL) + + h := &recordHandler{} + event.Register("foo", h) + event.Register("bar", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _ = event.Push(context.Background(), "foo.exact", nil) + _, _ = event.Push(context.Background(), "foo.other", nil) + _, _ = event.Push(context.Background(), "bar.thing", nil) + + time.Sleep(100 * time.Millisecond) + + allEvents := allL.getEvents() + fooEvents := fooL.getEvents() + exactEvents := exactL.getEvents() + + if len(allEvents) != 3 { + t.Fatalf("all listener expected 3, got %d", len(allEvents)) + } + if len(fooEvents) != 2 { + t.Fatalf("foo.* listener expected 2, got %d", len(fooEvents)) + } + if len(exactEvents) != 1 { + t.Fatalf("foo.exact listener expected 1, got %d", len(exactEvents)) + } + if exactEvents[0].Type != "foo.exact" { + t.Fatalf("expected foo.exact, got %s", exactEvents[0].Type) + } +} + +func TestListener_Filter(t *testing.T) { + event.Reset() + defer event.Reset() + + filtered := &collectListener{} + event.Listen("foo.*", filtered, event.Filter(func(ev *types.Event) bool { + return ev.Type == "foo.keep" + })) + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _ = event.Push(context.Background(), "foo.keep", nil) + _, _ = event.Push(context.Background(), "foo.drop", nil) + + time.Sleep(100 * time.Millisecond) + events := filtered.getEvents() + if len(events) != 1 || events[0].Type != "foo.keep" { + t.Fatalf("filter should only pass foo.keep, got %v", events) + } +} + +func TestListener_BufferFull_Skip(t *testing.T) { + event.Reset() + defer event.Reset() + + // Use buffer size 2, listener that blocks + blocking := &blockingListener{unblock: make(chan struct{})} + event.Listen("foo.*", blocking, event.BufferSize(2)) + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + // Push 5 events; 1 being processed + 2 buffered = 3, rest skipped + for i := 0; i < 5; i++ { + _, _ = event.Push(context.Background(), "foo.item", i) + } + + time.Sleep(50 * time.Millisecond) + close(blocking.unblock) // unblock listener + time.Sleep(100 * time.Millisecond) + + count := blocking.count.Load() + if count > 3 { + t.Fatalf("expected at most 3 events with buffer=2, got %d", count) + } + if count < 1 { + t.Fatal("expected at least 1 event") + } +} + +type blockingListener struct { + unblock chan struct{} + count atomic.Int32 +} + +func (l *blockingListener) OnEvent(ev *types.Event) { + <-l.unblock + l.count.Add(1) +} + +func (l *blockingListener) Shutdown(ctx context.Context) error { return nil } + +func TestListener_Shutdown(t *testing.T) { + event.Reset() + defer event.Reset() + + listener := &collectListener{} + event.Listen("foo.*", listener) + event.Register("foo", &recordHandler{}) + _ = event.Start() + _ = event.Stop(context.Background()) + + if !listener.shut { + t.Fatal("listener Shutdown should have been called") + } +} + +func TestListener_PanicRecovery(t *testing.T) { + event.Reset() + defer event.Reset() + + var afterPanic atomic.Int32 + pl := &panicListener{afterPanic: &afterPanic} + event.Listen("foo.*", pl) + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _ = event.Push(context.Background(), "foo.panic", nil) + _, _ = event.Push(context.Background(), "foo.ok", nil) + time.Sleep(100 * time.Millisecond) + + if afterPanic.Load() < 1 { + t.Fatal("listener should recover from panic and process next event") + } +} + +type panicListener struct { + afterPanic *atomic.Int32 + first atomic.Bool +} + +func (l *panicListener) OnEvent(ev *types.Event) { + if !l.first.Load() { + l.first.Store(true) + panic("listener panic") + } + l.afterPanic.Add(1) +} + +func (l *panicListener) Shutdown(ctx context.Context) error { return nil } + +// --- Coverage: notify when listener manager not started --- + +func TestListener_NotifyBeforeStart(t *testing.T) { + event.Reset() + defer event.Reset() + + listener := &collectListener{} + event.Listen("foo.*", listener) + + // Register handler but do NOT start service; Push will fail with ErrNotStarted. + // Instead, we test that listener.notify returns silently before start. + event.Register("foo", &recordHandler{}) + + // Manually start and immediately stop to verify no events leaked + _ = event.Start() + _ = event.Stop(context.Background()) + + events := listener.getEvents() + if len(events) != 0 { + t.Fatalf("expected 0 events before any push, got %d", len(events)) + } +} diff --git a/event/option.go b/event/option.go new file mode 100644 index 00000000..ff578318 --- /dev/null +++ b/event/option.go @@ -0,0 +1,51 @@ +package event + +import "github.com/yaoapp/yao/event/types" + +// MaxWorkers sets the max concurrent worker goroutines for a Handler. +// Default is 512. Workers are fire-and-forget (goroutine ends after task). +func MaxWorkers(n int) types.HandlerOption { + return func(e *types.HandlerEntry) { + e.MaxWorkers = n + } +} + +// ReservedWorkers sets the number of workers reserved for Call events. +// Default is 10. Push can use MaxWorkers - Reserved; Call can use MaxWorkers. +func ReservedWorkers(n int) types.HandlerOption { + return func(e *types.HandlerEntry) { + e.ReservedWorkers = n + } +} + +// QueueSize sets the per-queue capacity. Default is 8192. +// When a queue is full, Push/Call returns ErrQueueFull immediately. +func QueueSize(n int) types.HandlerOption { + return func(e *types.HandlerEntry) { + e.QueueSize = n + } +} + +// Queue sets the queue key for a Push/Call invocation. +// Events with the same queue key are processed serially (FIFO). +func Queue(key string) types.PushOption { + return func(ev *types.Event) { + ev.Queue = key + } +} + +// Filter sets a custom filter function for Listen or Subscribe. +// Events that do not pass the filter are skipped. +func Filter(fn func(*types.Event) bool) types.FilterOption { + return func(e *types.FilterEntry) { + e.Filter = fn + } +} + +// BufferSize sets the Listener channel buffer size. Default is 8192. +// Only effective for Listen; ignored by Subscribe. +func BufferSize(n int) types.FilterOption { + return func(e *types.FilterEntry) { + e.BufferSize = n + } +} diff --git a/event/queue.go b/event/queue.go new file mode 100644 index 00000000..676d98dc --- /dev/null +++ b/event/queue.go @@ -0,0 +1,208 @@ +package event + +import ( + "context" + "sync" + + "github.com/yaoapp/yao/event/types" +) + +// queueItem wraps an event with its execution context and response channel. +type queueItem struct { + ctx context.Context + ev *types.Event + resp chan<- types.Result +} + +// eventQueue is a single FIFO queue bound to a specific handler prefix. +// Events are enqueued and consumed serially by a dedicated goroutine. +type eventQueue struct { + id string + prefix string + ch chan queueItem + released bool + aborted bool + mu sync.Mutex + done chan struct{} // closed when consumer goroutine exits +} + +// enqueue adds an event to the queue. Returns error if full, released, or aborted. +// The send to q.ch is performed while holding q.mu to prevent a race with +// release()/abort() closing the channel between the flag check and the send. +func (q *eventQueue) enqueue(ctx context.Context, ev *types.Event, resp chan<- types.Result) error { + q.mu.Lock() + defer q.mu.Unlock() + + if q.released || q.aborted { + return ErrQueueReleased + } + + select { + case q.ch <- queueItem{ctx: ctx, ev: ev, resp: resp}: + return nil + default: + return ErrQueueFull + } +} + +// release gracefully stops the queue: rejects new events, drains existing ones. +func (q *eventQueue) release() { + q.mu.Lock() + if q.released || q.aborted { + q.mu.Unlock() + return + } + q.released = true + close(q.ch) + q.mu.Unlock() +} + +// abort forcefully stops the queue: rejects new events, discards pending. +// The consumer goroutine detects the aborted flag and skips remaining items. +func (q *eventQueue) abort() { + q.mu.Lock() + if q.aborted { + q.mu.Unlock() + return + } + wasReleased := q.released + q.aborted = true + q.released = true + if !wasReleased { + close(q.ch) + } + q.mu.Unlock() +} + +// consumer is the goroutine that processes queued events serially. +func (q *eventQueue) consumer(pool *workerPool) { + defer close(q.done) + for item := range q.ch { + q.mu.Lock() + aborted := q.aborted + q.mu.Unlock() + if aborted { + continue + } + + // For Push events, use a non-cancellable context so that queued + // fire-and-forget events are not dropped when the caller's ctx expires. + // For Call events, preserve the caller's ctx for deadline/cancellation. + dispatchCtx := item.ctx + if !item.ev.IsCall { + dispatchCtx = context.WithoutCancel(item.ctx) + } + + done, err := pool.dispatch(dispatchCtx, item.ev, item.resp) + if err != nil { + select { + case item.resp <- types.Result{Err: err}: + default: + } + continue + } + <-done + } +} + +// queueManager manages all active queues. +type queueManager struct { + mu sync.RWMutex + queues map[string]*eventQueue + released map[string]struct{} // tracks IDs that have been released/aborted +} + +func newQueueManager() *queueManager { + return &queueManager{ + queues: make(map[string]*eventQueue), + released: make(map[string]struct{}), + } +} + +// create creates a new queue bound to a handler prefix. +func (qm *queueManager) create(prefix string, queueID string, queueSize int, pool *workerPool) error { + qm.mu.Lock() + defer qm.mu.Unlock() + + if _, exists := qm.queues[queueID]; exists { + return ErrQueueExists + } + + q := &eventQueue{ + id: queueID, + prefix: prefix, + ch: make(chan queueItem, queueSize), + done: make(chan struct{}), + } + qm.queues[queueID] = q + go q.consumer(pool) + return nil +} + +// get returns a queue by ID. +// Returns ErrQueueNotFound if the queue was never created, +// or ErrQueueReleased if it has been released/aborted. +func (qm *queueManager) get(queueID string) (*eventQueue, error) { + qm.mu.RLock() + defer qm.mu.RUnlock() + + q, ok := qm.queues[queueID] + if !ok { + if _, wasReleased := qm.released[queueID]; wasReleased { + return nil, ErrQueueReleased + } + return nil, ErrQueueNotFound + } + return q, nil +} + +// release gracefully releases a queue. +func (qm *queueManager) release(queueID string) { + qm.mu.Lock() + q, ok := qm.queues[queueID] + if !ok { + qm.mu.Unlock() + return + } + delete(qm.queues, queueID) + qm.released[queueID] = struct{}{} + qm.mu.Unlock() + + q.release() + go func() { <-q.done }() +} + +// abortOne forcefully releases a single queue. +func (qm *queueManager) abortOne(queueID string) { + qm.mu.Lock() + q, ok := qm.queues[queueID] + if !ok { + qm.mu.Unlock() + return + } + delete(qm.queues, queueID) + qm.released[queueID] = struct{}{} + qm.mu.Unlock() + + q.abort() + go func() { <-q.done }() +} + +// abortAll forcefully releases all queues. Used during Stop. +func (qm *queueManager) abortAll() { + qm.mu.Lock() + queues := make([]*eventQueue, 0, len(qm.queues)) + for id, q := range qm.queues { + queues = append(queues, q) + qm.released[id] = struct{}{} + } + qm.queues = make(map[string]*eventQueue) + qm.mu.Unlock() + + for _, q := range queues { + q.abort() + } + for _, q := range queues { + <-q.done + } +} diff --git a/event/queue_test.go b/event/queue_test.go new file mode 100644 index 00000000..e2028eed --- /dev/null +++ b/event/queue_test.go @@ -0,0 +1,387 @@ +package event_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --- Phase 4: Queue tests --- + +// orderHandler records the order of payload values to verify FIFO. +type orderHandler struct { + mu sync.Mutex + order []int +} + +func (h *orderHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + var v int + if err := ev.Should(&v); err == nil { + h.mu.Lock() + h.order = append(h.order, v) + h.mu.Unlock() + } + if ev.IsCall { + resp <- types.Result{Data: v} + } +} + +func (h *orderHandler) Shutdown(ctx context.Context) error { return nil } + +func (h *orderHandler) getOrder() []int { + h.mu.Lock() + defer h.mu.Unlock() + cp := make([]int, len(h.order)) + copy(cp, h.order) + return cp +} + +func TestQueueCreate_Release_FIFO(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &orderHandler{} + event.Register("seq", h, event.QueueSize(100)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, err := event.QueueCreate("seq") + if err != nil { + t.Fatalf("QueueCreate failed: %v", err) + } + if qID == "" { + t.Fatal("expected non-empty queue ID") + } + + n := 20 + for i := 0; i < n; i++ { + _, err := event.Push(context.Background(), "seq.append", i, event.Queue(qID)) + if err != nil { + t.Fatalf("Push %d failed: %v", i, err) + } + } + + // Release and wait for drain + event.QueueRelease(qID) + time.Sleep(200 * time.Millisecond) + + order := h.getOrder() + if len(order) != n { + t.Fatalf("expected %d events, got %d", n, len(order)) + } + for i, v := range order { + if v != i { + t.Fatalf("FIFO violation at index %d: expected %d, got %d", i, i, v) + } + } +} + +func TestQueueCreate_CustomID(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, err := event.QueueCreate("seq", "my-custom-id") + if err != nil { + t.Fatalf("QueueCreate failed: %v", err) + } + if qID != "my-custom-id" { + t.Fatalf("expected my-custom-id, got %s", qID) + } + event.QueueRelease(qID) +} + +func TestQueueCreate_Duplicate(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, _ = event.QueueCreate("seq", "dup-id") + _, err := event.QueueCreate("seq", "dup-id") + if err != event.ErrQueueExists { + t.Fatalf("expected ErrQueueExists, got %v", err) + } + event.QueueRelease("dup-id") +} + +func TestQueueCreate_UnregisteredPrefix(t *testing.T) { + event.Reset() + defer event.Reset() + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, err := event.QueueCreate("nonexist") + if err != event.ErrNoHandler { + t.Fatalf("expected ErrNoHandler, got %v", err) + } +} + +func TestPush_QueueNotFound(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, err := event.Push(context.Background(), "seq.append", 1, event.Queue("no-such-queue")) + if err != event.ErrQueueNotFound { + t.Fatalf("expected ErrQueueNotFound, got %v", err) + } +} + +func TestPush_QueueReleased(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, _ := event.QueueCreate("seq") + event.QueueRelease(qID) + time.Sleep(50 * time.Millisecond) + + _, err := event.Push(context.Background(), "seq.append", 1, event.Queue(qID)) + if err != event.ErrQueueReleased { + t.Fatalf("expected ErrQueueReleased after release, got %v", err) + } +} + +func TestQueueAbort_DiscardsPending(t *testing.T) { + event.Reset() + defer event.Reset() + + // slowHandler delays processing to let events pile up + var processed atomic.Int32 + slow := &slowHandler{delay: 50 * time.Millisecond, counter: &processed} + event.Register("slow", slow, event.QueueSize(100)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, _ := event.QueueCreate("slow") + + // Push 10 events; first will start processing, rest queue up + for i := 0; i < 10; i++ { + _, _ = event.Push(context.Background(), "slow.work", i, event.Queue(qID)) + } + + time.Sleep(30 * time.Millisecond) // let first event start + event.QueueAbort(qID) + time.Sleep(200 * time.Millisecond) + + count := processed.Load() + if count >= 10 { + t.Fatalf("abort should discard pending events, but %d were processed", count) + } +} + +// slowHandler processes events with a delay. +type slowHandler struct { + delay time.Duration + counter *atomic.Int32 +} + +func (h *slowHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + time.Sleep(h.delay) + h.counter.Add(1) + if ev.IsCall { + resp <- types.Result{Data: "done"} + } +} + +func (h *slowHandler) Shutdown(ctx context.Context) error { return nil } + +func TestQueue_CallInsideQueue_Serial(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &orderHandler{} + event.Register("seq", h, event.QueueSize(100)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, _ := event.QueueCreate("seq") + defer event.QueueRelease(qID) + + // Push 5, then Call, then Push 5 more + for i := 0; i < 5; i++ { + _, _ = event.Push(context.Background(), "seq.append", i, event.Queue(qID)) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, data, err := event.Call(ctx, "seq.append", 99, event.Queue(qID)) + if err != nil { + t.Fatalf("Call failed: %v", err) + } + if data != 99 { + t.Fatalf("expected 99, got %v", data) + } + + for i := 5; i < 10; i++ { + _, _ = event.Push(context.Background(), "seq.append", i, event.Queue(qID)) + } + + time.Sleep(200 * time.Millisecond) + order := h.getOrder() + + // The Call (99) should appear after the first 5 and before the last 5 + found := false + for i, v := range order { + if v == 99 { + if i < 5 { + t.Fatalf("Call should be after first 5 pushes, found at index %d", i) + } + found = true + break + } + } + if !found { + t.Fatalf("Call result (99) not found in order: %v", order) + } +} + +func TestQueueFull(t *testing.T) { + event.Reset() + defer event.Reset() + + var processed atomic.Int32 + slow := &slowHandler{delay: 100 * time.Millisecond, counter: &processed} + event.Register("tiny", slow, event.QueueSize(2)) + _ = event.Start() + + qID, _ := event.QueueCreate("tiny") + + // Fill the queue (size=2) + _, err1 := event.Push(context.Background(), "tiny.work", 1, event.Queue(qID)) + _, err2 := event.Push(context.Background(), "tiny.work", 2, event.Queue(qID)) + + // These may or may not succeed depending on timing, but eventually one should fail + var fullErr error + for i := 0; i < 10; i++ { + _, err := event.Push(context.Background(), "tiny.work", i+3, event.Queue(qID)) + if err == event.ErrQueueFull { + fullErr = err + break + } + } + + if err1 != nil { + t.Fatalf("first push should succeed: %v", err1) + } + if err2 != nil { + t.Fatalf("second push should succeed: %v", err2) + } + if fullErr == nil { + t.Log("warning: queue never reported full (handler may be too fast)") + } + + // Wait for queued events to finish before Stop to avoid race between + // consumer goroutine (dispatch/wg.Add) and Stop (pool.wait/wg.Wait). + event.QueueRelease(qID) + time.Sleep(500 * time.Millisecond) + _ = event.Stop(context.Background()) +} + +// --- Coverage: QueueRelease idempotent (release non-existent queue) --- + +func TestQueueRelease_NonExistent(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + // Should not panic + event.QueueRelease("never-created") +} + +// --- Coverage: QueueAbort idempotent (abort non-existent queue) --- + +func TestQueueAbort_NonExistent(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + // Should not panic + event.QueueAbort("never-created") +} + +// --- Coverage: QueueAbort after already released --- + +func TestQueueAbort_AfterRelease(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, _ := event.QueueCreate("seq") + event.QueueRelease(qID) + time.Sleep(50 * time.Millisecond) + + // Abort after release should not panic (already removed from map) + event.QueueAbort(qID) +} + +// --- Coverage: Stop with active queues (abortAll path) --- + +func TestStop_WithActiveQueues(t *testing.T) { + event.Reset() + defer event.Reset() + + var processed atomic.Int32 + slow := &slowHandler{delay: 30 * time.Millisecond, counter: &processed} + event.Register("bg", slow, event.QueueSize(100)) + _ = event.Start() + + qID, _ := event.QueueCreate("bg") + for i := 0; i < 5; i++ { + _, _ = event.Push(context.Background(), "bg.work", i, event.Queue(qID)) + } + time.Sleep(10 * time.Millisecond) + + // Stop should abort all queues and wait + err := event.Stop(context.Background()) + if err != nil { + t.Fatalf("Stop failed: %v", err) + } +} + +// --- Coverage: Call with queue enqueue failure (queue released) --- + +func TestCall_QueueReleased(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("seq", &orderHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + qID, _ := event.QueueCreate("seq") + event.QueueRelease(qID) + time.Sleep(50 * time.Millisecond) + + _, _, err := event.Call(context.Background(), "seq.get", nil, event.Queue(qID)) + if err != event.ErrQueueReleased { + t.Fatalf("expected ErrQueueReleased, got %v", err) + } +} diff --git a/event/service.go b/event/service.go new file mode 100644 index 00000000..d9dfc7a2 --- /dev/null +++ b/event/service.go @@ -0,0 +1,221 @@ +package event + +import ( + "context" + "errors" + "sync" + + "github.com/yaoapp/yao/event/types" +) + +// Sentinel errors. +var ( + ErrNotStarted = errors.New("event: service not started") + ErrAlreadyStart = errors.New("event: service already started") + ErrQueueFull = errors.New("event: queue is full") + ErrQueueNotFound = errors.New("event: queue not found") + ErrQueueExists = errors.New("event: queue already exists") + ErrQueueReleased = errors.New("event: queue already released") + ErrNoHandler = errors.New("event: no handler registered for prefix") + ErrHandlerPanic = errors.New("event: handler panicked") +) + +// Context keys for SID and Auth propagation. +type ctxKey int + +const ( + ctxKeySID ctxKey = iota + ctxKeyAuth +) + +// WithSID returns a context carrying the given session ID. +func WithSID(ctx context.Context, sid string) context.Context { + return context.WithValue(ctx, ctxKeySID, sid) +} + +// SIDFrom extracts the session ID from ctx. Returns empty string if not set. +func SIDFrom(ctx context.Context) string { + if v, ok := ctx.Value(ctxKeySID).(string); ok { + return v + } + return "" +} + +// WithAuth returns a context carrying the given authorized info. +func WithAuth(ctx context.Context, auth *types.AuthorizedInfo) context.Context { + return context.WithValue(ctx, ctxKeyAuth, auth) +} + +// AuthFrom extracts the authorized info from ctx. Returns nil if not set. +func AuthFrom(ctx context.Context) *types.AuthorizedInfo { + if v, ok := ctx.Value(ctxKeyAuth).(*types.AuthorizedInfo); ok { + return v + } + return nil +} + +// service holds all global state for the event bus. +type service struct { + mu sync.RWMutex + started bool + handlers map[string]*types.HandlerEntry // prefix -> registration + pools map[string]*workerPool // prefix -> worker pool + queues *queueManager // queue lifecycle + lmgr *listenerManager // listener manager + smgr *subManager // subscriber manager +} + +var svc = &service{} + +func init() { + svc.reset() +} + +// Register registers a handler for the given prefix. +// Must be called before Start (typically in init()). +func Register(prefix string, handler types.Handler, opts ...types.HandlerOption) { + entry := &types.HandlerEntry{ + Prefix: prefix, + Handler: handler, + MaxWorkers: types.DefaultMaxWorkers, + ReservedWorkers: types.DefaultReservedWorkers, + QueueSize: types.DefaultQueueSize, + } + for _, opt := range opts { + opt(entry) + } + + svc.mu.Lock() + defer svc.mu.Unlock() + svc.handlers[prefix] = entry +} + +// Start initializes and starts the event service. +// Called during engine startup, after runtime is ready. +func Start() error { + svc.mu.Lock() + defer svc.mu.Unlock() + + if svc.started { + return ErrAlreadyStart + } + + // Create worker pools for each registered handler + for prefix, entry := range svc.handlers { + svc.pools[prefix] = newWorkerPool(entry) + } + + // Start listener manager + svc.lmgr.start() + + svc.started = true + return nil +} + +// Stop gracefully shuts down the event service. +// Waits for in-flight events to finish, discards pending queue items, +// and calls Shutdown on all handlers and listeners. +// +// The lock is released before waiting for workers so that in-flight handlers +// calling Push/Call (which acquire RLock via getHandler) do not deadlock. +// Once started=false, getHandler returns ErrNotStarted for any new calls. +func Stop(ctx context.Context) error { + svc.mu.Lock() + if !svc.started { + svc.mu.Unlock() + return nil + } + svc.started = false + + // Snapshot references under lock, then release. + queues := svc.queues + pools := make([]*workerPool, 0, len(svc.pools)) + for _, p := range svc.pools { + pools = append(pools, p) + } + handlers := make([]*types.HandlerEntry, 0, len(svc.handlers)) + for _, e := range svc.handlers { + handlers = append(handlers, e) + } + lmgr := svc.lmgr + smgr := svc.smgr + svc.mu.Unlock() + + // From here on, started=false prevents any new Push/Call/QueueCreate. + // Existing in-flight workers may still call getHandler and get ErrNotStarted, + // which is the correct behavior during shutdown. + + // Abort all queues (discard pending, wait for in-flight) + queues.abortAll() + + // Wait for all worker pools to drain + for _, pool := range pools { + pool.wait() + } + + // Shutdown all handlers + for _, entry := range handlers { + if entry.Handler != nil { + _ = entry.Handler.Shutdown(ctx) + } + } + + // Stop listener manager + lmgr.stop(ctx) + + // Clear subscribers + smgr.clear() + + return nil +} + +// Reload performs a hot-reload. Preserves queues and in-flight events, +// reloads dynamic configuration only. +func Reload() error { + svc.mu.RLock() + defer svc.mu.RUnlock() + + if !svc.started { + return ErrNotStarted + } + return nil +} + +// IsStarted reports whether the service is currently running. +func IsStarted() bool { + svc.mu.RLock() + defer svc.mu.RUnlock() + return svc.started +} + +// getHandler returns the handler entry and its worker pool for the given prefix. +func getHandler(prefix string) (*types.HandlerEntry, *workerPool, error) { + svc.mu.RLock() + defer svc.mu.RUnlock() + + if !svc.started { + return nil, nil, ErrNotStarted + } + entry, ok := svc.handlers[prefix] + if !ok { + return nil, nil, ErrNoHandler + } + pool := svc.pools[prefix] + return entry, pool, nil +} + +// Reset clears all state. For testing only. +func Reset() { + svc.mu.Lock() + defer svc.mu.Unlock() + svc.reset() +} + +func (s *service) reset() { + s.started = false + s.handlers = make(map[string]*types.HandlerEntry) + s.pools = make(map[string]*workerPool) + s.queues = newQueueManager() + s.lmgr = newListenerManager() + s.smgr = newSubManager() +} diff --git a/event/service_test.go b/event/service_test.go new file mode 100644 index 00000000..9a8b7196 --- /dev/null +++ b/event/service_test.go @@ -0,0 +1,229 @@ +package event_test + +import ( + "context" + "testing" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// stubHandler is a minimal Handler for testing registration and lifecycle. +type stubHandler struct { + shutdownCalled bool +} + +func (h *stubHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) {} + +func (h *stubHandler) Shutdown(ctx context.Context) error { + h.shutdownCalled = true + return nil +} + +// --- Register + Start/Stop lifecycle --- + +func TestStartStop_Basic(t *testing.T) { + event.Reset() + defer event.Reset() + + if event.IsStarted() { + t.Fatal("service should not be started initially") + } + + if err := event.Start(); err != nil { + t.Fatalf("Start failed: %v", err) + } + if !event.IsStarted() { + t.Fatal("service should be started after Start") + } + + if err := event.Stop(context.Background()); err != nil { + t.Fatalf("Stop failed: %v", err) + } + if event.IsStarted() { + t.Fatal("service should not be started after Stop") + } +} + +func TestStart_Double(t *testing.T) { + event.Reset() + defer event.Reset() + + if err := event.Start(); err != nil { + t.Fatalf("Start failed: %v", err) + } + + err := event.Start() + if err != event.ErrAlreadyStart { + t.Fatalf("expected ErrAlreadyStart, got: %v", err) + } + + _ = event.Stop(context.Background()) +} + +func TestStop_WhenNotStarted(t *testing.T) { + event.Reset() + defer event.Reset() + + if err := event.Stop(context.Background()); err != nil { + t.Fatalf("Stop on non-started service should succeed, got: %v", err) + } +} + +func TestReload_WhenNotStarted(t *testing.T) { + event.Reset() + defer event.Reset() + + err := event.Reload() + if err != event.ErrNotStarted { + t.Fatalf("expected ErrNotStarted, got: %v", err) + } +} + +func TestReload_WhenStarted(t *testing.T) { + event.Reset() + defer event.Reset() + + _ = event.Start() + if err := event.Reload(); err != nil { + t.Fatalf("Reload failed: %v", err) + } + _ = event.Stop(context.Background()) +} + +// --- Register + options --- + +func TestRegister_DefaultOptions(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &stubHandler{} + event.Register("test", h) + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + if !event.IsStarted() { + t.Fatal("service should be started") + } +} + +func TestRegister_CustomOptions(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &stubHandler{} + event.Register("test", h, + event.MaxWorkers(128), + event.ReservedWorkers(5), + event.QueueSize(2048), + ) + + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + if !event.IsStarted() { + t.Fatal("service should be started") + } +} + +// --- Stop calls Shutdown on handlers --- + +func TestStop_CallsHandlerShutdown(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &stubHandler{} + event.Register("test", h) + _ = event.Start() + + if err := event.Stop(context.Background()); err != nil { + t.Fatalf("Stop failed: %v", err) + } + if !h.shutdownCalled { + t.Fatal("Handler.Shutdown should have been called on Stop") + } +} + +func TestStop_MultipleHandlersShutdown(t *testing.T) { + event.Reset() + defer event.Reset() + + h1 := &stubHandler{} + h2 := &stubHandler{} + event.Register("alpha", h1) + event.Register("bravo", h2) + _ = event.Start() + + if err := event.Stop(context.Background()); err != nil { + t.Fatalf("Stop failed: %v", err) + } + if !h1.shutdownCalled || !h2.shutdownCalled { + t.Fatal("all handlers should have been shut down") + } +} + +// --- Context SID/Auth propagation --- + +func TestWithSID_SIDFrom(t *testing.T) { + ctx := event.WithSID(context.Background(), "sess-123") + got := event.SIDFrom(ctx) + if got != "sess-123" { + t.Fatalf("expected sess-123, got %s", got) + } +} + +func TestSIDFrom_Empty(t *testing.T) { + got := event.SIDFrom(context.Background()) + if got != "" { + t.Fatalf("expected empty, got %s", got) + } +} + +func TestWithAuth_AuthFrom(t *testing.T) { + auth := &types.AuthorizedInfo{UserID: "u-1", TeamID: "t-1"} + ctx := event.WithAuth(context.Background(), auth) + got := event.AuthFrom(ctx) + if got == nil { + t.Fatal("expected non-nil auth") + } + if got.UserID != "u-1" || got.TeamID != "t-1" { + t.Fatalf("unexpected auth: %+v", got) + } +} + +func TestAuthFrom_Nil(t *testing.T) { + got := event.AuthFrom(context.Background()) + if got != nil { + t.Fatal("expected nil auth from bare context") + } +} + +func TestWithSIDAndAuth_Combined(t *testing.T) { + auth := &types.AuthorizedInfo{UserID: "u-2"} + ctx := event.WithSID(context.Background(), "sess-456") + ctx = event.WithAuth(ctx, auth) + + if event.SIDFrom(ctx) != "sess-456" { + t.Fatal("SID mismatch") + } + if event.AuthFrom(ctx).UserID != "u-2" { + t.Fatal("Auth mismatch") + } +} + +// --- Reset --- + +func TestReset_ClearsState(t *testing.T) { + event.Reset() + + h := &stubHandler{} + event.Register("test", h) + _ = event.Start() + + event.Reset() + + if event.IsStarted() { + t.Fatal("service should not be started after Reset") + } +} diff --git a/event/sub.go b/event/sub.go new file mode 100644 index 00000000..089ad94d --- /dev/null +++ b/event/sub.go @@ -0,0 +1,101 @@ +package event + +import ( + "fmt" + "sync" + "sync/atomic" + + "github.com/yaoapp/yao/event/types" +) + +var subIDCounter atomic.Uint64 + +func nextSubID() string { + id := subIDCounter.Add(1) + return fmt.Sprintf("sub-%d", id) +} + +// subEntry holds a dynamic subscriber registration. +type subEntry struct { + id string + pattern string + filter func(*types.Event) bool + ch chan<- *types.Event +} + +// subManager manages dynamic subscribers. +type subManager struct { + mu sync.RWMutex + entries map[string]*subEntry // id -> entry +} + +func newSubManager() *subManager { + return &subManager{ + entries: make(map[string]*subEntry), + } +} + +// subscribe adds a dynamic subscriber. Returns the subscription ID. +func (sm *subManager) subscribe(pattern string, ch chan<- *types.Event, opts ...types.FilterOption) string { + fe := &types.FilterEntry{Pattern: pattern} + for _, opt := range opts { + opt(fe) + } + + id := nextSubID() + sm.mu.Lock() + defer sm.mu.Unlock() + sm.entries[id] = &subEntry{ + id: id, + pattern: pattern, + filter: fe.Filter, + ch: ch, + } + return id +} + +// unsubscribe removes a subscriber by ID. +func (sm *subManager) unsubscribe(id string) { + sm.mu.Lock() + defer sm.mu.Unlock() + delete(sm.entries, id) +} + +// notify sends an event to all matching subscribers (non-blocking). +func (sm *subManager) notify(ev *types.Event) { + sm.mu.RLock() + defer sm.mu.RUnlock() + + for _, entry := range sm.entries { + if !matchPattern(entry.pattern, ev.Type) { + continue + } + if entry.filter != nil && !entry.filter(ev) { + continue + } + select { + case entry.ch <- ev: + default: + // Subscriber chan full, skip (non-blocking) + } + } +} + +// clear removes all subscribers. Used during Stop. +func (sm *subManager) clear() { + sm.mu.Lock() + defer sm.mu.Unlock() + sm.entries = make(map[string]*subEntry) +} + +// Subscribe dynamically subscribes to events matching the given pattern. +// Returns the subscription ID for later unsubscription. +// Event delivery is non-blocking: if ch is full, the event is skipped. +func Subscribe(pattern string, ch chan<- *types.Event, opts ...types.FilterOption) string { + return svc.smgr.subscribe(pattern, ch, opts...) +} + +// Unsubscribe removes a dynamic subscription by ID. +func Unsubscribe(id string) { + svc.smgr.unsubscribe(id) +} diff --git a/event/sub_test.go b/event/sub_test.go new file mode 100644 index 00000000..abdde4f9 --- /dev/null +++ b/event/sub_test.go @@ -0,0 +1,181 @@ +package event_test + +import ( + "context" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --- Phase 7: Subscriber tests --- + +func TestSubscribe_Basic(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ch := make(chan *types.Event, 10) + subID := event.Subscribe("foo.*", ch) + if subID == "" { + t.Fatal("expected non-empty subscription ID") + } + defer event.Unsubscribe(subID) + + _, _ = event.Push(context.Background(), "foo.bar", "payload") + _, _ = event.Push(context.Background(), "foo.baz", "payload2") + + received := drainChan(ch, 2, 200*time.Millisecond) + if len(received) != 2 { + t.Fatalf("expected 2 events, got %d", len(received)) + } + if received[0].Type != "foo.bar" { + t.Fatalf("expected foo.bar, got %s", received[0].Type) + } + if received[1].Type != "foo.baz" { + t.Fatalf("expected foo.baz, got %s", received[1].Type) + } +} + +func TestSubscribe_PatternFilter(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + event.Register("bar", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ch := make(chan *types.Event, 10) + subID := event.Subscribe("foo.*", ch, event.Filter(func(ev *types.Event) bool { + return ev.Type == "foo.keep" + })) + defer event.Unsubscribe(subID) + + _, _ = event.Push(context.Background(), "foo.keep", nil) + _, _ = event.Push(context.Background(), "foo.drop", nil) + _, _ = event.Push(context.Background(), "bar.thing", nil) + + received := drainChan(ch, 1, 200*time.Millisecond) + if len(received) != 1 { + t.Fatalf("expected 1 filtered event, got %d", len(received)) + } + if received[0].Type != "foo.keep" { + t.Fatalf("expected foo.keep, got %s", received[0].Type) + } +} + +func TestSubscribe_Unsubscribe(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ch := make(chan *types.Event, 10) + subID := event.Subscribe("foo.*", ch) + + _, _ = event.Push(context.Background(), "foo.first", nil) + time.Sleep(50 * time.Millisecond) + + event.Unsubscribe(subID) + + _, _ = event.Push(context.Background(), "foo.second", nil) + time.Sleep(50 * time.Millisecond) + + received := drainChan(ch, 10, 100*time.Millisecond) + for _, ev := range received { + if ev.Type == "foo.second" { + t.Fatal("should not receive events after Unsubscribe") + } + } +} + +func TestSubscribe_ChanFull_Skip(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ch := make(chan *types.Event, 1) // tiny buffer + subID := event.Subscribe("foo.*", ch) + defer event.Unsubscribe(subID) + + // Push multiple events quickly; only 1 should fit in buffer + for i := 0; i < 5; i++ { + _, _ = event.Push(context.Background(), "foo.item", i) + } + + time.Sleep(100 * time.Millisecond) + + // Should have at most 1 in channel (rest skipped) + count := len(ch) + if count > 1 { + t.Fatalf("expected at most 1 buffered event, got %d", count) + } +} + +func TestSubscribe_WildcardAll(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + event.Register("bar", &recordHandler{}) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + ch := make(chan *types.Event, 10) + subID := event.Subscribe("*", ch) + defer event.Unsubscribe(subID) + + _, _ = event.Push(context.Background(), "foo.one", nil) + _, _ = event.Push(context.Background(), "bar.two", nil) + + received := drainChan(ch, 2, 200*time.Millisecond) + if len(received) != 2 { + t.Fatalf("wildcard * should receive all events, got %d", len(received)) + } +} + +func TestSubscribe_StopClearsSubscribers(t *testing.T) { + event.Reset() + defer event.Reset() + + event.Register("foo", &recordHandler{}) + _ = event.Start() + + ch := make(chan *types.Event, 10) + _ = event.Subscribe("foo.*", ch) + + _ = event.Stop(context.Background()) + + // After Stop, Push should fail + _, err := event.Push(context.Background(), "foo.bar", nil) + if err != event.ErrNotStarted { + t.Fatalf("expected ErrNotStarted after Stop, got %v", err) + } +} + +// drainChan reads up to n events from ch within timeout. +func drainChan(ch chan *types.Event, n int, timeout time.Duration) []*types.Event { + var result []*types.Event + timer := time.NewTimer(timeout) + defer timer.Stop() + + for range n { + select { + case ev := <-ch: + result = append(result, ev) + case <-timer.C: + return result + } + } + return result +} diff --git a/event/types/interfaces.go b/event/types/interfaces.go new file mode 100644 index 00000000..9cf17294 --- /dev/null +++ b/event/types/interfaces.go @@ -0,0 +1,23 @@ +package types + +import "context" + +// Handler processes events for a given prefix (registered at startup, one per prefix). +// +// Handle is invoked by the WorkerPool. +// - ctx: for Call, this carries the caller's deadline/cancellation; for Push, a non-cancellable context. +// - resp is always non-nil. For Push the framework passes a discard channel; for Call it waits for a read. +// Use ev.IsCall to decide whether to write a meaningful result. +type Handler interface { + Handle(ctx context.Context, ev *Event, resp chan<- Result) + Shutdown(ctx context.Context) error +} + +// Listener receives matched events in a dedicated goroutine (registered at startup). +// +// OnEvent is called in the Listener's own goroutine; it does not block other +// Listeners or Subscribers. +type Listener interface { + OnEvent(ev *Event) + Shutdown(ctx context.Context) error +} diff --git a/event/types/types.go b/event/types/types.go new file mode 100644 index 00000000..aaed979a --- /dev/null +++ b/event/types/types.go @@ -0,0 +1,102 @@ +package types + +import ( + "fmt" + "reflect" + + "github.com/yaoapp/gou/process" +) + +// AuthorizedInfo is an alias for gou/process.AuthorizedInfo. +type AuthorizedInfo = process.AuthorizedInfo + +// Event represents a single event in the event bus. +type Event struct { + Type string // Event type, e.g. "trace.add", "job.progress" + ID string // Auto-generated event ID + Queue string // Queue key for serial processing; empty means no queue + IsCall bool // true = synchronous Call, false = asynchronous Push + Payload any // Business data; concrete type is determined by event type + SID string // Session ID, extracted from caller context + Auth *AuthorizedInfo // Authorized info, extracted from caller context; may be nil +} + +// Should asserts the Payload to the target pointer type. +// target must be a non-nil pointer. Returns an error if the type does not match. +// +// Usage: +// +// var p MyPayload +// if err := ev.Should(&p); err != nil { ... } +func (ev *Event) Should(target any) error { + if target == nil { + return fmt.Errorf("event.Should: target must be a non-nil pointer") + } + + rv := reflect.ValueOf(target) + if rv.Kind() != reflect.Ptr || rv.IsNil() { + return fmt.Errorf("event.Should: target must be a non-nil pointer, got %T", target) + } + + if ev.Payload == nil { + return fmt.Errorf("event.Should: payload is nil") + } + + // Direct assignment: payload is already the expected pointer type + payloadVal := reflect.ValueOf(ev.Payload) + targetElem := rv.Elem() + + // If payload is a pointer, dereference it + if payloadVal.Kind() == reflect.Ptr { + if payloadVal.IsNil() { + return fmt.Errorf("event.Should: payload is nil pointer") + } + payloadVal = payloadVal.Elem() + } + + if !payloadVal.Type().AssignableTo(targetElem.Type()) { + return fmt.Errorf("event.Should: payload type %T is not assignable to %s", ev.Payload, targetElem.Type()) + } + + targetElem.Set(payloadVal) + return nil +} + +// Result holds the response from a synchronous Call. +type Result struct { + Data any + Err error +} + +// HandlerOption configures a Handler registration. +type HandlerOption func(*HandlerEntry) + +// HandlerEntry is the internal registration record for a Handler. +type HandlerEntry struct { + Prefix string + Handler Handler + MaxWorkers int // Max concurrent workers, default 512 + ReservedWorkers int // Workers reserved for Call, default 10 + QueueSize int // Per-queue capacity, default 8192 +} + +// FilterOption configures a Listener or Subscriber registration. +type FilterOption func(*FilterEntry) + +// FilterEntry is the internal registration record for a Listener/Subscriber. +type FilterEntry struct { + Pattern string + Filter func(*Event) bool // Custom filter function + BufferSize int // Listener chan buffer size, default 8192; only for Listen +} + +// PushOption configures a Push or Call invocation. +type PushOption func(*Event) + +// Default configuration values. +const ( + DefaultMaxWorkers = 512 + DefaultReservedWorkers = 10 + DefaultQueueSize = 8192 + DefaultBufferSize = 8192 +) diff --git a/event/types/types_test.go b/event/types/types_test.go new file mode 100644 index 00000000..8fd99b5a --- /dev/null +++ b/event/types/types_test.go @@ -0,0 +1,267 @@ +package types_test + +import ( + "fmt" + "testing" + + "github.com/yaoapp/yao/event/types" +) + +// samplePayload is a test-only struct with no business semantics. +type samplePayload struct { + Name string + Value int + Tags []string +} + +// --- Should: basic struct assignment --- + +func TestShould_StructValue(t *testing.T) { + ev := &types.Event{ + Payload: samplePayload{Name: "alpha", Value: 1, Tags: []string{"a", "b"}}, + } + + var got samplePayload + if err := ev.Should(&got); err != nil { + t.Fatalf("Should returned error: %v", err) + } + if got.Name != "alpha" || got.Value != 1 || len(got.Tags) != 2 { + t.Fatalf("unexpected payload: %+v", got) + } +} + +// --- Should: pointer payload --- + +func TestShould_PointerPayload(t *testing.T) { + ev := &types.Event{ + Payload: &samplePayload{Name: "beta", Value: 2}, + } + + var got samplePayload + if err := ev.Should(&got); err != nil { + t.Fatalf("Should returned error: %v", err) + } + if got.Name != "beta" || got.Value != 2 { + t.Fatalf("unexpected payload: %+v", got) + } +} + +// --- Should: primitive payloads --- + +func TestShould_StringPayload(t *testing.T) { + ev := &types.Event{ + Payload: "hello world", + } + + var got string + if err := ev.Should(&got); err != nil { + t.Fatalf("Should returned error: %v", err) + } + if got != "hello world" { + t.Fatalf("unexpected string: %s", got) + } +} + +func TestShould_IntPayload(t *testing.T) { + ev := &types.Event{ + Payload: 42, + } + + var got int + if err := ev.Should(&got); err != nil { + t.Fatalf("Should returned error: %v", err) + } + if got != 42 { + t.Fatalf("unexpected int: %d", got) + } +} + +// --- Should: error cases --- + +func TestShould_NilTarget(t *testing.T) { + ev := &types.Event{Payload: "data"} + if err := ev.Should(nil); err == nil { + t.Fatal("expected error for nil target") + } +} + +func TestShould_NonPointerTarget(t *testing.T) { + ev := &types.Event{Payload: "data"} + var s string + if err := ev.Should(s); err == nil { + t.Fatal("expected error for non-pointer target") + } +} + +func TestShould_NilPayload(t *testing.T) { + ev := &types.Event{Payload: nil} + var got string + if err := ev.Should(&got); err == nil { + t.Fatal("expected error for nil payload") + } +} + +func TestShould_NilPointerPayload(t *testing.T) { + ev := &types.Event{Payload: (*samplePayload)(nil)} + var got samplePayload + if err := ev.Should(&got); err == nil { + t.Fatal("expected error for nil pointer payload") + } +} + +func TestShould_TypeMismatch(t *testing.T) { + ev := &types.Event{ + Payload: "wrong type", + } + var got samplePayload + if err := ev.Should(&got); err == nil { + t.Fatal("expected error for type mismatch") + } +} + +// --- Event fields --- + +func TestEvent_NilAuth(t *testing.T) { + ev := &types.Event{ + Type: "x.y", + ID: "ev-100", + Auth: nil, + } + if ev.Auth != nil { + t.Fatal("Auth should be nil") + } + if ev.Type != "x.y" || ev.ID != "ev-100" { + t.Fatalf("unexpected Type/ID: %s/%s", ev.Type, ev.ID) + } +} + +func TestEvent_WithAuth(t *testing.T) { + ev := &types.Event{ + Type: "x.y", + ID: "ev-101", + SID: "sess-abc", + Auth: &types.AuthorizedInfo{ + UserID: "u-1", + TeamID: "t-1", + }, + } + if ev.Type != "x.y" || ev.ID != "ev-101" { + t.Fatalf("unexpected Type/ID: %s/%s", ev.Type, ev.ID) + } + if ev.SID != "sess-abc" { + t.Fatalf("unexpected SID: %s", ev.SID) + } + if ev.Auth.UserID != "u-1" || ev.Auth.TeamID != "t-1" { + t.Fatalf("unexpected Auth: %+v", ev.Auth) + } +} + +func TestEvent_QueueAndIsCall(t *testing.T) { + push := &types.Event{Queue: "q-1", IsCall: false} + call := &types.Event{Queue: "q-1", IsCall: true} + + if push.IsCall { + t.Fatal("Push event should not be IsCall") + } + if !call.IsCall { + t.Fatal("Call event should be IsCall") + } + if push.Queue != "q-1" || call.Queue != "q-1" { + t.Fatal("Queue key mismatch") + } +} + +// --- Result --- + +func TestResult_Success(t *testing.T) { + r := types.Result{Data: map[string]string{"k": "v"}, Err: nil} + if r.Err != nil { + t.Fatal("expected nil error") + } + m, ok := r.Data.(map[string]string) + if !ok || m["k"] != "v" { + t.Fatal("unexpected result data") + } +} + +func TestResult_Error(t *testing.T) { + r := types.Result{Data: nil, Err: fmt.Errorf("something failed")} + if r.Err == nil { + t.Fatal("expected error") + } + if r.Err.Error() != "something failed" { + t.Fatalf("unexpected error message: %s", r.Err.Error()) + } + if r.Data != nil { + t.Fatal("expected nil data") + } +} + +// --- HandlerEntry defaults --- + +func TestHandlerEntry_Defaults(t *testing.T) { + entry := types.HandlerEntry{} + if entry.MaxWorkers != 0 { + t.Fatal("zero value should be 0 before applying options") + } + + if entry.MaxWorkers == 0 { + entry.MaxWorkers = types.DefaultMaxWorkers + } + if entry.ReservedWorkers == 0 { + entry.ReservedWorkers = types.DefaultReservedWorkers + } + if entry.QueueSize == 0 { + entry.QueueSize = types.DefaultQueueSize + } + + if entry.MaxWorkers != 512 { + t.Fatalf("expected MaxWorkers 512, got %d", entry.MaxWorkers) + } + if entry.ReservedWorkers != 10 { + t.Fatalf("expected ReservedWorkers 10, got %d", entry.ReservedWorkers) + } + if entry.QueueSize != 8192 { + t.Fatalf("expected QueueSize 8192, got %d", entry.QueueSize) + } +} + +// --- FilterEntry --- + +func TestFilterEntry_WithFilter(t *testing.T) { + called := false + entry := types.FilterEntry{ + Pattern: "x.*", + Filter: func(ev *types.Event) bool { + called = true + return ev.Type == "x.hit" + }, + BufferSize: 4096, + } + + if entry.Pattern != "x.*" { + t.Fatalf("unexpected Pattern: %s", entry.Pattern) + } + if !entry.Filter(&types.Event{Type: "x.hit"}) { + t.Fatal("filter should match x.hit") + } + if !called { + t.Fatal("filter was not called") + } + if entry.Filter(&types.Event{Type: "x.miss"}) { + t.Fatal("filter should not match x.miss") + } + if entry.BufferSize != 4096 { + t.Fatalf("unexpected BufferSize: %d", entry.BufferSize) + } +} + +func TestFilterEntry_NilFilter(t *testing.T) { + entry := types.FilterEntry{Pattern: "y.*"} + if entry.Pattern != "y.*" { + t.Fatalf("unexpected Pattern: %s", entry.Pattern) + } + if entry.Filter != nil { + t.Fatal("Filter should be nil when not set") + } +} diff --git a/event/worker.go b/event/worker.go new file mode 100644 index 00000000..3fb752b8 --- /dev/null +++ b/event/worker.go @@ -0,0 +1,94 @@ +package event + +import ( + "context" + "sync" + + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/event/types" +) + +// workerPool manages goroutine-based workers for a single Handler. +// Workers are fire-and-forget: each goroutine processes one task then exits. +// MaxWorkers limits total concurrent goroutines. +// ReservedWorkers reserves slots for Call events so Push cannot starve them. +type workerPool struct { + handler types.Handler + + // semTotal is a buffered channel of size MaxWorkers. + semTotal chan struct{} + + // semPush is a buffered channel of size (MaxWorkers - ReservedWorkers). + // Push events must acquire from both semPush and semTotal. + // Call events only acquire from semTotal. + semPush chan struct{} + + wg sync.WaitGroup +} + +func newWorkerPool(entry *types.HandlerEntry) *workerPool { + pushSlots := entry.MaxWorkers - entry.ReservedWorkers + if pushSlots < 1 { + pushSlots = 1 + } + return &workerPool{ + handler: entry.Handler, + semTotal: make(chan struct{}, entry.MaxWorkers), + semPush: make(chan struct{}, pushSlots), + } +} + +// dispatch runs the handler for one event in a new goroutine. +// Returns a done channel that is closed when the handler finishes. +// Blocks until a worker slot is available or ctx is cancelled. +func (wp *workerPool) dispatch(ctx context.Context, ev *types.Event, resp chan<- types.Result) (done <-chan struct{}, err error) { + isPush := !ev.IsCall + + if isPush { + select { + case wp.semPush <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + + select { + case wp.semTotal <- struct{}{}: + case <-ctx.Done(): + if isPush { + <-wp.semPush + } + return nil, ctx.Err() + } + + ch := make(chan struct{}) + wp.wg.Add(1) + go func() { + defer close(ch) + defer wp.wg.Done() + defer func() { <-wp.semTotal }() + if isPush { + defer func() { <-wp.semPush }() + } + defer wp.recoverPanic(ev, resp) + + wp.handler.Handle(ctx, ev, resp) + }() + + return ch, nil +} + +func (wp *workerPool) recoverPanic(ev *types.Event, resp chan<- types.Result) { + if r := recover(); r != nil { + log.Error("event worker panic: type=%s id=%s err=%v", ev.Type, ev.ID, r) + select { + case resp <- types.Result{Err: ErrHandlerPanic}: + default: + } + } +} + +// wait blocks until all active workers finish. Used during Stop. +func (wp *workerPool) wait() { + wp.wg.Wait() +} diff --git a/event/worker_test.go b/event/worker_test.go new file mode 100644 index 00000000..d2106bf6 --- /dev/null +++ b/event/worker_test.go @@ -0,0 +1,213 @@ +package event_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/event/types" +) + +// --- Phase 5: Worker pool tests --- + +func TestWorker_MaxConcurrency(t *testing.T) { + event.Reset() + defer event.Reset() + + var peak atomic.Int32 + var current atomic.Int32 + + h := &concurrencyHandler{peak: &peak, current: ¤t, delay: 30 * time.Millisecond} + event.Register("conc", h, event.MaxWorkers(4), event.ReservedWorkers(1)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, _ = event.Push(context.Background(), "conc.work", i) + }(i) + } + wg.Wait() + time.Sleep(300 * time.Millisecond) + + p := peak.Load() + if p > 4 { + t.Fatalf("peak concurrency %d exceeded MaxWorkers 4", p) + } + if p < 2 { + t.Fatalf("peak concurrency %d seems too low, expected at least 2", p) + } +} + +type concurrencyHandler struct { + peak *atomic.Int32 + current *atomic.Int32 + delay time.Duration +} + +func (h *concurrencyHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + c := h.current.Add(1) + for { + old := h.peak.Load() + if c <= old || h.peak.CompareAndSwap(old, c) { + break + } + } + time.Sleep(h.delay) + h.current.Add(-1) + if ev.IsCall { + resp <- types.Result{Data: "ok"} + } +} + +func (h *concurrencyHandler) Shutdown(ctx context.Context) error { return nil } + +func TestWorker_CallReservation(t *testing.T) { + event.Reset() + defer event.Reset() + + // MaxWorkers=4, ReservedWorkers=2 => Push can use 2, Call can use 4 + var pushActive atomic.Int32 + var callDone atomic.Int32 + + h := &reservationHandler{pushActive: &pushActive, callDone: &callDone} + event.Register("res", h, event.MaxWorkers(4), event.ReservedWorkers(2)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + // Saturate push slots (only 2 available for push) + for i := 0; i < 4; i++ { + _, _ = event.Push(context.Background(), "res.work", i) + } + time.Sleep(20 * time.Millisecond) // let pushes start + + // Call should still work (reserved slots) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, data, err := event.Call(ctx, "res.get", "ping") + if err != nil { + t.Fatalf("Call should succeed with reserved workers: %v", err) + } + if data != "pong" { + t.Fatalf("expected pong, got %v", data) + } +} + +type reservationHandler struct { + pushActive *atomic.Int32 + callDone *atomic.Int32 +} + +func (h *reservationHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + if ev.IsCall { + resp <- types.Result{Data: "pong"} + h.callDone.Add(1) + return + } + h.pushActive.Add(1) + time.Sleep(100 * time.Millisecond) + h.pushActive.Add(-1) +} + +func (h *reservationHandler) Shutdown(ctx context.Context) error { return nil } + +func TestWorker_PanicRecovery(t *testing.T) { + event.Reset() + defer event.Reset() + + var afterPanic atomic.Bool + + h := &panicHandler{afterPanic: &afterPanic} + event.Register("pan", h) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + // First push panics + _, _ = event.Push(context.Background(), "pan.crash", "boom") + time.Sleep(50 * time.Millisecond) + + // Second push should still work + _, _ = event.Push(context.Background(), "pan.ok", "fine") + time.Sleep(50 * time.Millisecond) + + if !afterPanic.Load() { + t.Fatal("handler should have processed event after panic recovery") + } +} + +type panicHandler struct { + afterPanic *atomic.Bool +} + +func (h *panicHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { + if ev.Type == "pan.crash" { + panic("test panic") + } + h.afterPanic.Store(true) + if ev.IsCall { + resp <- types.Result{Data: "ok"} + } +} + +func (h *panicHandler) Shutdown(ctx context.Context) error { return nil } + +// --- Coverage: ReservedWorkers >= MaxWorkers (pushSlots clamped to 1) --- + +func TestWorker_ReservedExceedsMax(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &recordHandler{} + event.Register("edge", h, event.MaxWorkers(2), event.ReservedWorkers(5)) + _ = event.Start() + defer func() { _ = event.Stop(context.Background()) }() + + _, err := event.Push(context.Background(), "edge.work", "data") + if err != nil { + t.Fatalf("Push failed: %v", err) + } + time.Sleep(50 * time.Millisecond) + + calls := h.getCalls() + if len(calls) != 1 { + t.Fatalf("expected 1 call, got %d", len(calls)) + } +} + +// --- Coverage: dispatch Call ctx cancel while waiting for semTotal --- + +func TestWorker_Call_CtxCancel_SemTotal(t *testing.T) { + event.Reset() + defer event.Reset() + + h := &concurrencyHandler{ + peak: &atomic.Int32{}, + current: &atomic.Int32{}, + delay: 200 * time.Millisecond, + } + event.Register("lim", h, event.MaxWorkers(1), event.ReservedWorkers(0)) + _ = event.Start() + + bgDone := make(chan struct{}) + go func() { + defer close(bgDone) + _, _, _ = event.Call(context.Background(), "lim.work", nil) + }() + time.Sleep(10 * time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + _, _, err := event.Call(ctx, "lim.op", nil) + if err == nil { + t.Fatal("expected error for call with saturated pool") + } + + <-bgDone + _ = event.Stop(context.Background()) +}