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.
This commit is contained in:
Max 2026-02-23 12:18:13 +08:00
parent ccb1a9404c
commit fd24e31912
21 changed files with 3843 additions and 8 deletions

2
.gitignore vendored
View file

@ -66,3 +66,5 @@ release/*
sandbox/TODO-VNC.md
sandbox/docker/chrome/PLAN.md
sandbox/DESIGN-REMOTE.md
event/DESIGN.md
event/TODO.md

View file

@ -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 ""

184
event/README.md Normal file
View file

@ -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 MaxReserved) |
| `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
```

382
event/bench_test.go Normal file
View file

@ -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++
}
})
}

159
event/bus.go Normal file
View file

@ -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)
}

320
event/bus_test.go Normal file
View file

@ -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())
}

348
event/leak_test.go Normal file
View file

@ -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)
}
}

141
event/listener.go Normal file
View file

@ -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...)
}

222
event/listener_test.go Normal file
View file

@ -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))
}
}

51
event/option.go Normal file
View file

@ -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
}
}

208
event/queue.go Normal file
View file

@ -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
}
}

387
event/queue_test.go Normal file
View file

@ -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)
}
}

221
event/service.go Normal file
View file

@ -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()
}

229
event/service_test.go Normal file
View file

@ -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")
}
}

101
event/sub.go Normal file
View file

@ -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)
}

181
event/sub_test.go Normal file
View file

@ -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
}

23
event/types/interfaces.go Normal file
View file

@ -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
}

102
event/types/types.go Normal file
View file

@ -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
)

267
event/types/types_test.go Normal file
View file

@ -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")
}
}

94
event/worker.go Normal file
View file

@ -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()
}

213
event/worker_test.go Normal file
View file

@ -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: &current, 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())
}