1824dbd feat(app): route internal capture replies · neferwl

A internal/app/captured.go +41 -0

@@ -0,0 +1,41 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+
	6	+	"github.com/bnema/neferwl/internal/adapters/logging"
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// routeCaptured demultiplexes capture replies from the outputs: replies whose
	11	+	// ID carries ports.CaptureIDInternal go to internal, all others to wayland.
	12	+	// Both sends give way to ctx. A nil internal means no internal consumer is
	13	+	// wired: internal replies are dropped (logged at debug) instead of blocking
	14	+	// forever, since nothing would ever receive them. It returns when ctx is done
	15	+	// or in is closed.
	16	+	func routeCaptured(ctx context.Context, in <-chan ports.CaptureDone, wayland, internal chan<- ports.CaptureDone) {
	17	+	log := logging.For(ctx, "app")
	18	+	for {
	19	+	select {
	20	+	case <-ctx.Done():
	21	+	return
	22	+	case done, ok := <-in:
	23	+	if !ok {
	24	+	return
	25	+	}
	26	+	dst := wayland
	27	+	if done.ID&ports.CaptureIDInternal != 0 {
	28	+	if internal == nil {
	29	+	log.Debug().Uint64("id", done.ID).Msg("internal capture reply dropped: no consumer")
	30	+	continue
	31	+	}
	32	+	dst = internal
	33	+	}
	34	+	select {
	35	+	case dst <- done:
	36	+	case <-ctx.Done():
	37	+	return
	38	+	}
	39	+	}
	40	+	}
	41	+	}

@@ -0,0 +1,41 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+
	6	+	"github.com/bnema/neferwl/internal/adapters/logging"
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// routeCaptured demultiplexes capture replies from the outputs: replies whose
	11	+	// ID carries ports.CaptureIDInternal go to internal, all others to wayland.
	12	+	// Both sends give way to ctx. A nil internal means no internal consumer is
	13	+	// wired: internal replies are dropped (logged at debug) instead of blocking
	14	+	// forever, since nothing would ever receive them. It returns when ctx is done
	15	+	// or in is closed.
	16	+	func routeCaptured(ctx context.Context, in <-chan ports.CaptureDone, wayland, internal chan<- ports.CaptureDone) {
	17	+	log := logging.For(ctx, "app")
	18	+	for {
	19	+	select {
	20	+	case <-ctx.Done():
	21	+	return
	22	+	case done, ok := <-in:
	23	+	if !ok {
	24	+	return
	25	+	}
	26	+	dst := wayland
	27	+	if done.ID&ports.CaptureIDInternal != 0 {
	28	+	if internal == nil {
	29	+	log.Debug().Uint64("id", done.ID).Msg("internal capture reply dropped: no consumer")
	30	+	continue
	31	+	}
	32	+	dst = internal
	33	+	}
	34	+	select {
	35	+	case dst <- done:
	36	+	case <-ctx.Done():
	37	+	return
	38	+	}
	39	+	}
	40	+	}
	41	+	}

A internal/app/captured_test.go +72 -0

@@ -0,0 +1,72 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/stretchr/testify/require"
	9	+
	10	+	"github.com/bnema/neferwl/internal/ports"
	11	+	)
	12	+
	13	+	func TestCapturedDemux(t *testing.T) {
	14	+	ctx, cancel := context.WithCancel(context.Background())
	15	+	defer cancel()
	16	+	in := make(chan ports.CaptureDone, 4)
	17	+	wl := make(chan ports.CaptureDone, 4)
	18	+	internal := make(chan ports.CaptureDone, 4)
	19	+	exited := make(chan struct{})
	20	+	go func() { routeCaptured(ctx, in, wl, internal); close(exited) }()
	21	+
	22	+	in <- ports.CaptureDone{ID: 7}
	23	+	in <- ports.CaptureDone{ID: ports.CaptureIDInternal \| 3}
	24	+	require.Equal(t, uint64(7), recvDone(t, wl).ID)
	25	+	require.Equal(t, ports.CaptureIDInternal\|3, recvDone(t, internal).ID)
	26	+	require.Empty(t, wl)
	27	+	require.Empty(t, internal)
	28	+
	29	+	cancel()
	30	+	select {
	31	+	case <-exited:
	32	+	case <-time.After(2 * time.Second):
	33	+	t.Fatal("routeCaptured did not return on cancel")
	34	+	}
	35	+	}
	36	+
	37	+	func TestCapturedDemuxCancelWhileBlocked(t *testing.T) {
	38	+	ctx, cancel := context.WithCancel(context.Background())
	39	+	in := make(chan ports.CaptureDone, 1)
	40	+	wl := make(chan ports.CaptureDone) // never read
	41	+	exited := make(chan struct{})
	42	+	go func() { routeCaptured(ctx, in, wl, nil); close(exited) }()
	43	+	in <- ports.CaptureDone{ID: 1}
	44	+	cancel()
	45	+	select {
	46	+	case <-exited:
	47	+	case <-time.After(2 * time.Second):
	48	+	t.Fatal("routeCaptured blocked on a full destination after cancel")
	49	+	}
	50	+	}
	51	+
	52	+	func TestCapturedDemuxDropsInternalWithoutConsumer(t *testing.T) {
	53	+	ctx, cancel := context.WithCancel(context.Background())
	54	+	defer cancel()
	55	+	in := make(chan ports.CaptureDone, 2)
	56	+	wl := make(chan ports.CaptureDone, 2)
	57	+	go routeCaptured(ctx, in, wl, nil)
	58	+	in <- ports.CaptureDone{ID: ports.CaptureIDInternal \| 1}
	59	+	in <- ports.CaptureDone{ID: 2}
	60	+	require.Equal(t, uint64(2), recvDone(t, wl).ID)
	61	+	}
	62	+
	63	+	func recvDone(t *testing.T, ch <-chan ports.CaptureDone) ports.CaptureDone {
	64	+	t.Helper()
	65	+	select {
	66	+	case d := <-ch:
	67	+	return d
	68	+	case <-time.After(2 * time.Second):
	69	+	t.Fatal("no capture reply")
	70	+	return ports.CaptureDone{}
	71	+	}
	72	+	}

@@ -0,0 +1,72 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/stretchr/testify/require"
	9	+
	10	+	"github.com/bnema/neferwl/internal/ports"
	11	+	)
	12	+
	13	+	func TestCapturedDemux(t *testing.T) {
	14	+	ctx, cancel := context.WithCancel(context.Background())
	15	+	defer cancel()
	16	+	in := make(chan ports.CaptureDone, 4)
	17	+	wl := make(chan ports.CaptureDone, 4)
	18	+	internal := make(chan ports.CaptureDone, 4)
	19	+	exited := make(chan struct{})
	20	+	go func() { routeCaptured(ctx, in, wl, internal); close(exited) }()
	21	+
	22	+	in <- ports.CaptureDone{ID: 7}
	23	+	in <- ports.CaptureDone{ID: ports.CaptureIDInternal \| 3}
	24	+	require.Equal(t, uint64(7), recvDone(t, wl).ID)
	25	+	require.Equal(t, ports.CaptureIDInternal\|3, recvDone(t, internal).ID)
	26	+	require.Empty(t, wl)
	27	+	require.Empty(t, internal)
	28	+
	29	+	cancel()
	30	+	select {
	31	+	case <-exited:
	32	+	case <-time.After(2 * time.Second):
	33	+	t.Fatal("routeCaptured did not return on cancel")
	34	+	}
	35	+	}
	36	+
	37	+	func TestCapturedDemuxCancelWhileBlocked(t *testing.T) {
	38	+	ctx, cancel := context.WithCancel(context.Background())
	39	+	in := make(chan ports.CaptureDone, 1)
	40	+	wl := make(chan ports.CaptureDone) // never read
	41	+	exited := make(chan struct{})
	42	+	go func() { routeCaptured(ctx, in, wl, nil); close(exited) }()
	43	+	in <- ports.CaptureDone{ID: 1}
	44	+	cancel()
	45	+	select {
	46	+	case <-exited:
	47	+	case <-time.After(2 * time.Second):
	48	+	t.Fatal("routeCaptured blocked on a full destination after cancel")
	49	+	}
	50	+	}
	51	+
	52	+	func TestCapturedDemuxDropsInternalWithoutConsumer(t *testing.T) {
	53	+	ctx, cancel := context.WithCancel(context.Background())
	54	+	defer cancel()
	55	+	in := make(chan ports.CaptureDone, 2)
	56	+	wl := make(chan ports.CaptureDone, 2)
	57	+	go routeCaptured(ctx, in, wl, nil)
	58	+	in <- ports.CaptureDone{ID: ports.CaptureIDInternal \| 1}
	59	+	in <- ports.CaptureDone{ID: 2}
	60	+	require.Equal(t, uint64(2), recvDone(t, wl).ID)
	61	+	}
	62	+
	63	+	func recvDone(t *testing.T, ch <-chan ports.CaptureDone) ports.CaptureDone {
	64	+	t.Helper()
	65	+	select {
	66	+	case d := <-ch:
	67	+	return d
	68	+	case <-time.After(2 * time.Second):
	69	+	t.Fatal("no capture reply")
	70	+	return ports.CaptureDone{}
	71	+	}
	72	+	}

M internal/app/session.go +2 -0


@@ -340,6 +340,8 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
340	340		log.Warn().Err(err).Msg("capture allowlist not watched, restart to apply changes")
341	341		}
342	342		}()
	343	+	workers.Add(1)
	344	+	go func() { defer workers.Done(); routeCaptured(ctx, p.capturedAll, p.captured, p.internalCaptured()) }()
343	345		go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
344	346		go func() { defer workers.Done(); done <- c.Run(ctx) }()
345	347		workers.Add(1)


@@ -340,6 +340,8 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
340	log.Warn().Err(err).Msg("capture allowlist not watched, restart to apply changes")	340		log.Warn().Err(err).Msg("capture allowlist not watched, restart to apply changes")
341	}	341		}
342	}()	342		}()
		343	+	workers.Add(1)
		344	+	go func() { defer workers.Done(); routeCaptured(ctx, p.capturedAll, p.captured, p.internalCaptured()) }()
343	go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()	345		go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
344	go func() { defer workers.Done(); done <- c.Run(ctx) }()	346		go func() { defer workers.Done(); done <- c.Run(ctx) }()
345	workers.Add(1)	347		workers.Add(1)

M internal/app/wiring.go +22 -6


@@ -47,9 +47,17 @@ type pipes struct {
47	47		frames chan ports.OutputFrame
48	48
49	49		captures chan ports.CaptureRequest
50		-	// Every admitted capture owns one reply slot until wayland consumes it.
51		-	// Keep this capacity tied to admission so output routing never waits.
52		-	captured chan ports.CaptureDone
	50	+	// Every admitted capture owns one reply slot until routeCaptured takes
	51	+	// it. Keep this capacity tied to admission so output routing never waits.
	52	+	// The outputs reply on capturedAll; routeCaptured splits the replies by
	53	+	// ports.CaptureIDInternal into captured (the wayland consumer) and
	54	+	// hintCaptured (internal consumers).
	55	+	capturedAll chan ports.CaptureDone
	56	+	captured chan ports.CaptureDone
	57	+	// hintCaptured carries internal capture replies. Until an internal
	58	+	// consumer (the hints adapter) is wired, nothing reads it: see
	59	+	// internalCaptured.
	60	+	hintCaptured chan ports.CaptureDone
53	61
54	62		outputFormats chan ports.OutputFormats
55	63		outputHeads chan ports.OutputHeads

@@ -97,8 +105,10 @@ func newPipes(realSession bool) *pipes {
97	105
98	106		frames: make(chan ports.OutputFrame, 8),
99	107
100		-	captures: make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
101		-	captured: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
	108	+	captures: make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
	109	+	capturedAll: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
	110	+	captured: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
	111	+	hintCaptured: make(chan ports.CaptureDone, 1),
102	112
103	113		outputFormats: make(chan ports.OutputFormats, 8),
104	114		outputHeads: make(chan ports.OutputHeads, 8),

@@ -139,6 +149,12 @@ func (p *pipes) core() core.Channels {
139	149		}
140	150		}
141	151
	152	+	// internalCaptured is the channel internal capture replies are routed to, nil
	153	+	// while no internal consumer is wired: routeCaptured then drops them (debug
	154	+	// log) rather than blocking on a channel nobody reads. The hints adapter
	155	+	// makes it p.hintCaptured.
	156	+	func (p *pipes) internalCaptured() chan<- ports.CaptureDone { return nil }
	157	+
142	158		// wayland is the wayland server's side.
143	159		func (p *pipes) wayland() wayland.Channels {
144	160		return wayland.Channels{

@@ -175,7 +191,7 @@ func (p *pipes) outputs(security ports.SessionSecurity, presented ports.Presenta
175	191		content: content,
176	192		frames: p.frames,
177	193		captures: p.captures,
178		-	captured: p.captured,
	194	+	captured: p.capturedAll,
179	195		formats: p.outputFormats,
180	196		heads: p.outputHeads,
181	197		leaseRequests: p.leaseRequests,

 	frames chan ports.OutputFrame
 	captures chan ports.CaptureRequest
-	// Every admitted capture owns one reply slot until wayland consumes it.
+	// Every admitted capture owns one reply slot until routeCaptured takes
-	// Keep this capacity tied to admission so output routing never waits.
+	// it. Keep this capacity tied to admission so output routing never waits.
-	captured chan ports.CaptureDone
+	// The outputs reply on capturedAll; routeCaptured splits the replies by
+	// ports.CaptureIDInternal into captured (the wayland consumer) and
+	// hintCaptured (internal consumers).
+	capturedAll chan ports.CaptureDone
+	captured    chan ports.CaptureDone
+	// hintCaptured carries internal capture replies. Until an internal
+	// consumer (the hints adapter) is wired, nothing reads it: see
+	// internalCaptured.
+	hintCaptured chan ports.CaptureDone
 	outputFormats chan ports.OutputFormats
 	outputHeads   chan ports.OutputHeads
 		frames: make(chan ports.OutputFrame, 8),
-		captures: make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
+		captures:     make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
-		captured: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
+		capturedAll:  make(chan ports.CaptureDone, ports.MaxCaptureInflight),
+		captured:     make(chan ports.CaptureDone, ports.MaxCaptureInflight),
+		hintCaptured: make(chan ports.CaptureDone, 1),
 		outputFormats: make(chan ports.OutputFormats, 8),
 		outputHeads:   make(chan ports.OutputHeads, 8),
 	}
 }
+// internalCaptured is the channel internal capture replies are routed to, nil
+// while no internal consumer is wired: routeCaptured then drops them (debug
+// log) rather than blocking on a channel nobody reads. The hints adapter
+// makes it p.hintCaptured.
+func (p *pipes) internalCaptured() chan<- ports.CaptureDone { return nil }
 // wayland is the wayland server's side.
 func (p *pipes) wayland() wayland.Channels {
 	return wayland.Channels{
 		content:         content,
 		frames:          p.frames,
 		captures:        p.captures,
-		captured:        p.captured,
+		captured:        p.capturedAll,
 		formats:         p.outputFormats,
 		heads:           p.outputHeads,
 		leaseRequests:   p.leaseRequests,