diff --git a/cmd/olliesrv/internal/session/session_test.go b/cmd/olliesrv/internal/session/session_test.go index 40e8180..7f80e78 100644 --- a/cmd/olliesrv/internal/session/session_test.go +++ b/cmd/olliesrv/internal/session/session_test.go @@ -1,8 +1,11 @@ package session import ( + "context" + "fmt" "os" "testing" + "time" ) func TestSetCwd_ExpandsTilde(t *testing.T) { @@ -41,3 +44,77 @@ func TestSetCwd_AbsolutePathUnchanged(t *testing.T) { t.Errorf("SetCwd(/usr/local/bin): got %q, want %q", got, want) } } + +func TestEventSubscription(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch := SubscribeEvents(ctx, "*") + + // Give subscription time to register + time.Sleep(10 * time.Millisecond) + + // Publish events + for i := 0; i < 5; i++ { + PublishEvent(fmt.Sprintf("test.event.%d", i), fmt.Sprintf("payload%d", i)) + } + + // Collect events + timeout := time.After(1 * time.Second) + var received []Event + for { + select { + case ev := <-ch: + received = append(received, ev) + if len(received) >= 5 { + goto done + } + case <-timeout: + goto done + } + } +done: + if len(received) < 5 { + t.Errorf("Expected 5 events, got %d", len(received)) + for i, ev := range received { + t.Logf(" [%d] topic=%s payload=%s", i, ev.Topic, ev.Payload) + } + } +} + +func TestEventSubscriptionFiltered(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch := SubscribeEventsFiltered(ctx, "*") + + // Give subscription time to register + time.Sleep(10 * time.Millisecond) + + // Publish events + for i := 0; i < 5; i++ { + PublishEvent(fmt.Sprintf("test.event.%d", i), fmt.Sprintf("payload%d", i)) + } + + // Collect events + timeout := time.After(1 * time.Second) + var received []Event + for { + select { + case ev := <-ch: + received = append(received, ev) + if len(received) >= 5 { + goto done + } + case <-timeout: + goto done + } + } +done: + if len(received) < 5 { + t.Errorf("Expected 5 events, got %d", len(received)) + for i, ev := range received { + t.Logf(" [%d] topic=%s payload=%s", i, ev.Topic, ev.Payload) + } + } +} diff --git a/cmd/olliesrv/server.go b/cmd/olliesrv/server.go index ce367bf..88d31ea 100644 --- a/cmd/olliesrv/server.go +++ b/cmd/olliesrv/server.go @@ -602,6 +602,7 @@ func read(root *fs.Tree, cs *connState, fc *plan9.Fcall, ctx context.Context) *p if path == "event" { cs.mu.RLock() f, fidOK := cs.fids[fc.Fid] + connCtx := cs.ctx // Use connection context, not request context cs.mu.RUnlock() if fidOK { f.mu.Lock() @@ -614,7 +615,7 @@ func read(root *fs.Tree, cs *connState, fc *plan9.Fcall, ctx context.Context) *p if filter == "" { filter = "*" } - subCtx, cancel := context.WithCancel(ctx) + subCtx, cancel := context.WithCancel(connCtx) ch := session.SubscribeEventsFiltered(subCtx, filter) f.mu.Lock() f.eventFilter = filter