e0bbcba refactor(outputkit): Requests.Stop uses the Init context; tighten capture and lease tests · neferwl

M internal/adapters/drm/output.go +1 -1


@@ -1056,7 +1056,7 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (ports.Rend
1056	1056		defer func() {
1057	1057		cancelCaptures()
1058	1058		pipeline.Close(r)
1059		-	requests.Stop(ctx, captures)
	1059	+	requests.Stop(captures)
1060	1060		}()
1061	1061		clk := o.clock
1062	1062		if clk == nil {


@@ -1056,7 +1056,7 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (ports.Rend
1056		defer func() {	1056		defer func() {
1057		cancelCaptures()	1057		cancelCaptures()
1058		pipeline.Close(r)	1058		pipeline.Close(r)
1059	-	requests.Stop(ctx, captures)	1059	+	requests.Stop(captures)
1060		}()	1060		}()
1061		clk := o.clock	1061		clk := o.clock
1062		if clk == nil {	1062		if clk == nil {

M internal/adapters/headless/output.go +1 -1


@@ -105,7 +105,7 @@ func Run(ctx context.Context, opts Options, scenes <-chan ports.Scene, contents
105	105		pipeline.EndGate(capture.ErrFrameNotPresented)
106	106		cancelCaptures()
107	107		pipeline.Close(r)
108		-	requests.Stop(ctx, incoming)
	108	+	requests.Stop(incoming)
109	109		}()
110	110		var want ports.CursorChange
111	111		cursorScale := -1.0 // not loaded yet


@@ -105,7 +105,7 @@ func Run(ctx context.Context, opts Options, scenes <-chan ports.Scene, contents
105		pipeline.EndGate(capture.ErrFrameNotPresented)	105		pipeline.EndGate(capture.ErrFrameNotPresented)
106		cancelCaptures()	106		cancelCaptures()
107		pipeline.Close(r)	107		pipeline.Close(r)
108	-	requests.Stop(ctx, incoming)	108	+	requests.Stop(incoming)
109		}()	109		}()
110		var want ports.CursorChange	110		var want ports.CursorChange
111		cursorScale := -1.0 // not loaded yet	111		cursorScale := -1.0 // not loaded yet

M internal/adapters/outputkit/capture/requests.go +6 -6


@@ -28,7 +28,8 @@ type Requests struct {
28	28		replies chan<- ports.CaptureDone
29	29		}
30	30
31		-	// Init binds the requests to the owner's context and reply channel.
	31	+	// Init binds the requests to the owner's context and reply channel. It must
	32	+	// run before any other method.
32	33		func (r *Requests) Init(ctx context.Context, replies chan<- ports.CaptureDone) {
33	34		r.List, r.ctx, r.replies = r.storage[:0], ctx, replies
34	35		}

@@ -73,12 +74,11 @@ func (r *Requests) Reset() {
73	74		}
74	75
75	76		// Stop fails what the owner still holds, then what is queued on incoming,
76		-	// with ErrOutputStopped. ctx is the context replies are sent under. It
77		-	// returns once incoming is empty or closed.
78		-	func (r *Requests) Stop(ctx context.Context, incoming <-chan ports.CaptureRequest) {
	77	+	// with ErrOutputStopped. It returns once incoming is empty or closed.
	78	+	func (r *Requests) Stop(incoming <-chan ports.CaptureRequest) {
79	79		for _, q := range r.List {
80	80		if !Handed(q) {
81		-	Fail(ctx, q, ErrOutputStopped, r.replies)
	81	+	Fail(r.ctx, q, ErrOutputStopped, r.replies)
82	82		}
83	83		}
84	84		for {

@@ -87,7 +87,7 @@ func (r *Requests) Stop(ctx context.Context, incoming <-chan ports.CaptureReques
87	87		if !ok {
88	88		return
89	89		}
90		-	Fail(ctx, q, ErrOutputStopped, r.replies)
	90	+	Fail(r.ctx, q, ErrOutputStopped, r.replies)
91	91		default:
92	92		return
93	93		}

 	replies chan<- ports.CaptureDone
 }
-// Init binds the requests to the owner's context and reply channel.
+// Init binds the requests to the owner's context and reply channel. It must
+// run before any other method.
 func (r *Requests) Init(ctx context.Context, replies chan<- ports.CaptureDone) {
 	r.List, r.ctx, r.replies = r.storage[:0], ctx, replies
 }
 }
 // Stop fails what the owner still holds, then what is queued on incoming,
-// with ErrOutputStopped. ctx is the context replies are sent under. It
+// with ErrOutputStopped. It returns once incoming is empty or closed.
-// returns once incoming is empty or closed.
+func (r *Requests) Stop(incoming <-chan ports.CaptureRequest) {
-func (r *Requests) Stop(ctx context.Context, incoming <-chan ports.CaptureRequest) {
 	for _, q := range r.List {
 		if !Handed(q) {
-			Fail(ctx, q, ErrOutputStopped, r.replies)
+			Fail(r.ctx, q, ErrOutputStopped, r.replies)
 		}
 	}
 	for {
 			if !ok {
 				return
 			}
-			Fail(ctx, q, ErrOutputStopped, r.replies)
+			Fail(r.ctx, q, ErrOutputStopped, r.replies)
 		default:
 			return
 		}

M internal/adapters/outputkit/capture/requests_test.go +15 -3


@@ -52,12 +52,24 @@ func TestRequestsStopDrainsIncoming(t *testing.T) {
52	52		var r Requests
53	53		r.Init(context.Background(), replies)
54	54		r.Admit(ports.CaptureRequest{ID: 1}, nil, ports.Scene{}, false)
	55	+	r.Admit(ports.CaptureRequest{ID: 3}, nil, ports.Scene{}, false)
	56	+	r.List[1] = ports.CaptureRequest{} // handed to the pipeline
55	57		incoming := make(chan ports.CaptureRequest, 2)
56	58		incoming <- ports.CaptureRequest{ID: 2}
57	59		close(incoming)
58		-	r.Stop(context.Background(), incoming)
	60	+	r.Stop(incoming)
59	61		require.Len(t, replies, 2)
60		-	for range 2 {
61		-	require.ErrorIs(t, (<-replies).Err, ErrOutputStopped)
	62	+	for _, id := range []uint64{1, 2} {
	63	+	done := <-replies
	64	+	require.Equal(t, id, done.ID)
	65	+	require.ErrorIs(t, done.Err, ErrOutputStopped)
62	66		}
63	67		}
	68	+
	69	+	func TestRequestsStopReturnsOnOpenEmptyChannel(t *testing.T) {
	70	+	replies := make(chan ports.CaptureDone, 1)
	71	+	var r Requests
	72	+	r.Init(context.Background(), replies)
	73	+	r.Stop(make(chan ports.CaptureRequest))
	74	+	require.Empty(t, replies)
	75	+	}

 	var r Requests
 	r.Init(context.Background(), replies)
 	r.Admit(ports.CaptureRequest{ID: 1}, nil, ports.Scene{}, false)
+	r.Admit(ports.CaptureRequest{ID: 3}, nil, ports.Scene{}, false)
+	r.List[1] = ports.CaptureRequest{} // handed to the pipeline
 	incoming := make(chan ports.CaptureRequest, 2)
 	incoming <- ports.CaptureRequest{ID: 2}
 	close(incoming)
-	r.Stop(context.Background(), incoming)
+	r.Stop(incoming)
 	require.Len(t, replies, 2)
-	for range 2 {
+	for _, id := range []uint64{1, 2} {
-		require.ErrorIs(t, (<-replies).Err, ErrOutputStopped)
+		done := <-replies
+		require.Equal(t, id, done.ID)
+		require.ErrorIs(t, done.Err, ErrOutputStopped)
 	}
 }
+func TestRequestsStopReturnsOnOpenEmptyChannel(t *testing.T) {
+	replies := make(chan ports.CaptureDone, 1)
+	var r Requests
+	r.Init(context.Background(), replies)
+	r.Stop(make(chan ports.CaptureRequest))
+	require.Empty(t, replies)
+}

M internal/app/leases_test.go +1 -1


@@ -48,7 +48,7 @@ func TestLeasePublisherSendsChangesOnly(t *testing.T) {
48	48		t.Fatalf("inventory: %+v", inv)
49	49		}
50	50		defer inv.Device.Close()
51		-	if inv.Device == p.clientFDs[card] {
	51	+	if inv.Device.Fd() == p.clientFDs[card].Fd() {
52	52		t.Fatal("message shares the publisher's fd")
53	53		}
54	54		p.publish(card)


@@ -48,7 +48,7 @@ func TestLeasePublisherSendsChangesOnly(t *testing.T) {
48		t.Fatalf("inventory: %+v", inv)	48		t.Fatalf("inventory: %+v", inv)
49		}	49		}
50		defer inv.Device.Close()	50		defer inv.Device.Close()
51	-	if inv.Device == p.clientFDs[card] {	51	+	if inv.Device.Fd() == p.clientFDs[card].Fd() {
52		t.Fatal("message shares the publisher's fd")	52		t.Fatal("message shares the publisher's fd")
53		}	53		}
54		p.publish(card)	54		p.publish(card)