server: fix event subscription context - use connection ctx not request ctx
The subscription was being cancelled after the first read completed because we used the per-request context. Now using the connection context so the subscription lives until the fid is clunked or connection closed.
This commit is contained in:
parent
79ab165ffd
commit
43d3051a78
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue