refactor(app): make every session channel in newPipes and hand each side its ends (#103)

commit: 1c6cce962efffdf22ce0aa42351d92786a05739d
author: bnema · 19d ago · 2026-10-03 14:48 UTC
committer: GitHub · 19d ago · 2026-10-03 14:48 UTC
parent: 7e208c5
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

2 files · +239 -61

M internal/app/app.go +51 -61


@@ -88,46 +88,11 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
88	88		defer hw.close()
89	89		}
90	90		security := &sessionsecurity.Gate{}
91		-	securityChanges := make(chan ports.SecurityState, 8)
92		-	securityEvents := make(chan ports.SecurityBackendEvent, 64)
93		-	client := make(chan ports.ClientEvent, 32)
94		-	input := make(chan ports.InputEvent, 32)
95		-	output := make(chan ports.OutputEvent, 32)
96		-	configChanges := make(chan ports.ConfigChanged, 8)
97		-	commands := make(chan ports.ClientCommand, 32)
98		-	spawn := make(chan ports.SpawnRequest, 32)
99		-	scenes := make(chan []ports.Scene, 1)
100		-	layouts := make(chan ports.Layout, 1)
101		-	constraints := make(chan ports.PointerConstraint, 1)
102		-	states := make(chan ports.State, 1)
103		-	workspaces := make(chan ports.Workspaces, 1)
104		-	configErrors := make(chan error, 8)
105		-	renderScenes := make(chan []ports.Scene, 1)
106		-	contents := make(chan ports.SurfaceContent, 64)
107		-	cursorChanges := make(chan ports.CursorChange, 1)
108		-	presented := make(chan ports.OutputPresented, 64)
109		-	// Outputs report on flips; relayPresented passes them to wayland and
110		-	// tells core about each flip (frames) to move running slides.
111		-	flips := make(chan ports.OutputPresented, 64)
112		-	frames := make(chan ports.OutputFrame, 8)
113		-	captures := make(chan ports.CaptureRequest, ports.MaxCaptureInflight)
114		-	// Every admitted capture owns one reply slot until Wayland consumes it.
115		-	// Keep this capacity tied to admission so output routing never waits.
116		-	captured := make(chan ports.CaptureDone, ports.MaxCaptureInflight)
117		-	outputFormats := make(chan ports.OutputFormats, 8)
118		-	outputHeads := make(chan ports.OutputHeads, 8)
119		-	leaseRequests := make(chan ports.LeaseMessage, 32)
120		-	leaseEvents := make(chan ports.LeaseMessage, 32)
121		-	applyOutput := make(chan ports.OutputApply, 8)
122		-	appliedOutput := make(chan ports.OutputApplied, 8)
123		-	idleInhibited := make(chan bool, 1)
124		-	idleActivity := make(chan struct{}, 1)
125		-	var scales chan ports.ScaleChanged
126		-	if hw != nil {
127		-	// Only real sessions save scales: headless runs never touch the config file.
128		-	scales = make(chan ports.ScaleChanged, 8)
129		-	}
130		-	ch := core.Channels{Security: security, Scales: scales, Client: client, Input: input, Output: output, Config: configChanges, Commands: commands, Spawn: spawn, Scenes: scenes, Layouts: layouts, Constraints: constraints, State: states, Workspaces: workspaces, ConfigErrors: configErrors, Terminal: !opts.NoTerminal, Clock: clock.System{}, Frames: frames}
	91	+	p := newPipes(hw != nil)
	92	+	ch := p.core()
	93	+	ch.Security = security
	94	+	ch.Terminal = !opts.NoTerminal
	95	+	ch.Clock = clock.System{}
131	96		c, err := core.New(opts.Config, ch)
132	97		if err != nil {
133	98		return err

@@ -154,7 +119,17 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
154	119		} else {
155	120		captureAllow = captureallow.NewStore(captureallow.DefaultPath, captureAllowLog)
156	121		}
157		-	server, err := wayland.New(wayland.Options{CaptureAllow: captureAllow, WorkspaceIDs: wsIDs, Security: security, RuntimeDir: runtimeDir, DMABuf: dmabuf, SyncobjNode: renderNode(dmabuf.Device), Keymap: keymap, RepeatRate: opts.Config.Keyboard.RepeatRate, RepeatDelay: opts.Config.Keyboard.RepeatDelay}, wayland.Channels{SecurityChanges: securityChanges, SecurityEvents: securityEvents, Events: client, Commands: commands, Workspaces: workspaces, Contents: contents, Cursors: cursorChanges, Presented: presented, Captures: captures, Captured: captured, OutputFormats: outputFormats, OutputHeads: outputHeads, LeaseRequests: leaseRequests, LeaseEvents: leaseEvents, OutputApply: applyOutput, OutputApplied: appliedOutput, IdleInhibited: idleInhibited, IdleActivity: idleActivity}, logging.For(ctx, "wayland"))
	122	+	server, err := wayland.New(wayland.Options{
	123	+	CaptureAllow: captureAllow,
	124	+	WorkspaceIDs: wsIDs,
	125	+	Security: security,
	126	+	RuntimeDir: runtimeDir,
	127	+	DMABuf: dmabuf,
	128	+	SyncobjNode: renderNode(dmabuf.Device),
	129	+	Keymap: keymap,
	130	+	RepeatRate: opts.Config.Keyboard.RepeatRate,
	131	+	RepeatDelay: opts.Config.Keyboard.RepeatDelay,
	132	+	}, p.wayland(), logging.For(ctx, "wayland"))
158	133		if err != nil {
159	134		km.Close()
160	135		return err

@@ -180,7 +155,7 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
180	155		child := launcher.New(childEnv, logging.For(ctx, "launcher"))
181	156		child.Security = security
182	157		if inject != nil {
183		-	inject(input)
	158	+	inject(p.input)
184	159		}
185	160		var workers sync.WaitGroup
186	161		workers.Add(8)

@@ -189,27 +164,23 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
189	164		if path == "" {
190	165		path = config.DefaultPath()
191	166		}
192		-	watched := make(chan ports.ConfigChanged, 8)
193		-	keymaps := make(chan *xkb.Keymap, 1)
194		-	deviceConfigs := make(chan ports.InputDevicesConfig, 1)
195	167		go func() {
196	168		defer workers.Done()
197		-	done <- config.Watch(ctx, path, watched, logging.For(ctx, "config"))
	169	+	done <- config.Watch(ctx, path, p.watched, logging.For(ctx, "config"))
198	170		}()
199		-	if scales != nil {
	171	+	if p.scales != nil {
200	172		workers.Add(1)
201	173		go func() {
202	174		defer workers.Done()
203		-	core.PersistScales(ctx, scales, config.ScaleStore{Path: path}, launcher.NewNotifier(ctx, childEnv, logging.For(ctx, "launcher")), clock.System{}, time.Second)
	175	+	core.PersistScales(ctx, p.scales, config.ScaleStore{Path: path}, launcher.NewNotifier(ctx, childEnv, logging.For(ctx, "launcher")), clock.System{}, time.Second)
204	176		}()
205	177		}
206		-	filtered := make(chan ports.ConfigChanged, 8)
207	178		curs := newCursors(clock.System{}, opts.Config.Cursor.HideAfter)
208	179		defer curs.stop()
209	180		workers.Add(1)
210	181		go func() {
211	182		defer workers.Done()
212		-	relayConfig(ctx, opts.Config, watched, filtered, keymaps, deviceConfigs, curs, commands, logging.For(ctx, "config"))
	183	+	relayConfig(ctx, opts.Config, p.watched, p.filtered, p.keymaps, p.deviceConfigs, curs, p.commands, logging.For(ctx, "config"))
213	184		}()
214	185		script := make(chan string)
215	186		go func() {

@@ -218,7 +189,7 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
218	189		// Core sends the first layout once DRM reports the outputs.
219	190		var layout ports.Layout
220	191		select {
221		-	case layout = <-layouts:
	192	+	case layout = <-p.layouts:
222	193		case <-ctx.Done():
223	194		km.Close()
224	195		return

@@ -231,11 +202,27 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
231	202		log.Warn().Str("component", "sched").Err(err).Msg("input scheduling")
232	203		}
233	204		}
234		-	return libinput.Run(ctx, libinput.Options{Security: security, Seat: hw.seat, SeatName: hw.seat.Name(), Keymap: km, Keymaps: keymaps, Layout: layout, Layouts: layouts, Constraints: constraints, DeviceConfig: opts.Config.InputDevicesConfig, DeviceConfigs: deviceConfigs, Active: hw.seat.Subscribe(), MoveCursor: curs.move, Log: logging.For(ctx, "input"), LogMotion: logging.Enabled(ctx, "input-motion"), LogKeys: logging.Enabled(ctx, "input-keys")}, input)
	205	+	return libinput.Run(ctx, libinput.Options{
	206	+	Security: security,
	207	+	Seat: hw.seat,
	208	+	SeatName: hw.seat.Name(),
	209	+	Keymap: km,
	210	+	Keymaps: p.keymaps,
	211	+	Layout: layout,
	212	+	Layouts: p.layouts,
	213	+	Constraints: p.constraints,
	214	+	DeviceConfig: opts.Config.InputDevicesConfig,
	215	+	DeviceConfigs: p.deviceConfigs,
	216	+	Active: hw.seat.Subscribe(),
	217	+	MoveCursor: curs.move,
	218	+	Log: logging.For(ctx, "input"),
	219	+	LogMotion: logging.Enabled(ctx, "input-motion"),
	220	+	LogKeys: logging.Enabled(ctx, "input-keys"),
	221	+	}, p.input)
235	222		})
236	223		return
237	224		}
238		-	if err := headlessinput.RunSecure(ctx, km, keymaps, script, input, layouts, func(o string, x, y float64) { curs.move(o, x, y, true) }, logging.For(ctx, "input"), security); err != nil && !errors.Is(err, context.Canceled) {
	225	+	if err := headlessinput.RunSecure(ctx, km, p.keymaps, script, p.input, p.layouts, func(o string, x, y float64) { curs.move(o, x, y, true) }, logging.For(ctx, "input"), security); err != nil && !errors.Is(err, context.Canceled) {
239	226		select {
240	227		case done <- err:
241	228		case <-ctx.Done():

@@ -269,7 +256,7 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
269	256		log.Warn().Err(err).Msg("capture allowlist not watched, restart to apply changes")
270	257		}
271	258		}()
272		-	go func() { defer workers.Done(); done <- child.Run(ctx, spawn) }()
	259	+	go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
273	260		go func() { defer workers.Done(); done <- c.Run(ctx) }()
274	261		if xdisplay != nil {
275	262		workers.Add(1)

@@ -280,19 +267,22 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
280	267		workers.Add(1)
281	268		go func() {
282	269		defer workers.Done()
283		-	if err := statefile.Run(ctx, statePath, states, wsIDs, logging.For(ctx, "statefile")); err != nil {
	270	+	if err := statefile.Run(ctx, statePath, p.states, wsIDs, logging.For(ctx, "statefile")); err != nil {
284	271		// Scripts lose their state; the session goes on.
285	272		log.Warn().Err(err).Msg("state file disabled")
286	273		}
287	274		}()
288	275		}
289		-	go func() { defer workers.Done(); consumeScenes(ctx, scenes, configErrors, opts.testScenes, renderScenes) }()
	276	+	go func() {
	277	+	defer workers.Done()
	278	+	consumeScenes(ctx, p.scenes, p.configErrors, opts.testScenes, p.renderScenes)
	279	+	}()
290	280		workers.Add(1)
291		-	go func() { defer workers.Done(); relayPresented(ctx, flips, presented, frames) }()
	281	+	go func() { defer workers.Done(); relayPresented(ctx, p.flips, p.presented, p.frames) }()
292	282
293	283		go func() {
294	284		defer workers.Done()
295		-	outputIO := outputChannels{security: security, securityChanges: securityChanges, securityEvents: securityEvents, events: output, scenes: renderScenes, contents: contents, cursorChanges: cursorChanges, presented: flips, captures: captures, captured: captured, formats: outputFormats, heads: outputHeads, leaseRequests: leaseRequests, leaseEvents: leaseEvents, reloads: filtered, configured: configChanges, requests: applyOutput, replies: appliedOutput}
	285	+	outputIO := p.outputs(security)
296	286		apply := newOutputApply(newOutputOverrides(opts.Config, hw == nil), logging.For(ctx, "app"))
297	287		renderLog := logging.For(ctx, "render")
298	288		newRenderer := func(w, h int) (ports.Renderer, error) {

@@ -326,7 +316,7 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
326	316		workers.Add(1)
327	317		go func() {
328	318		defer workers.Done()
329		-	reports := screensaver.Reports{Held: idleInhibited, Activity: idleActivity}
	319	+	reports := screensaver.Reports{Held: p.idleInhibited, Activity: p.idleActivity}
330	320		screensaver.Serve(ctx, "", reports, screensaverRetry, logging.For(ctx, "screensaver"))
331	321		}()
332	322		}

@@ -342,10 +332,10 @@ func run(ctx context.Context, opts Options, inject func(chan<- ports.InputEvent)
342	332		cancel()
343	333		workers.Wait()
344	334		// The wayland server has stopped: no request can be queued any more.
345		-	drainCaptures(ctx, captures, nil)
	335	+	drainCaptures(ctx, p.captures, nil)
346	336		// A keymap the input goroutine never took is still ours to free.
347	337		select {
348		-	case km := <-keymaps:
	338	+	case km := <-p.keymaps:
349	339		km.Close()
350	340		default:
351	341		}

 		defer hw.close()
 	}
 	security := &sessionsecurity.Gate{}
-	securityChanges := make(chan ports.SecurityState, 8)
+	p := newPipes(hw != nil)
-	securityEvents := make(chan ports.SecurityBackendEvent, 64)
+	ch := p.core()
-	client := make(chan ports.ClientEvent, 32)
+	ch.Security = security
-	input := make(chan ports.InputEvent, 32)
+	ch.Terminal = !opts.NoTerminal
-	output := make(chan ports.OutputEvent, 32)
+	ch.Clock = clock.System{}
-	configChanges := make(chan ports.ConfigChanged, 8)
-	commands := make(chan ports.ClientCommand, 32)
-	spawn := make(chan ports.SpawnRequest, 32)
-	scenes := make(chan []ports.Scene, 1)
-	layouts := make(chan ports.Layout, 1)
-	constraints := make(chan ports.PointerConstraint, 1)
-	states := make(chan ports.State, 1)
-	workspaces := make(chan ports.Workspaces, 1)
-	configErrors := make(chan error, 8)
-	renderScenes := make(chan []ports.Scene, 1)
-	contents := make(chan ports.SurfaceContent, 64)
-	cursorChanges := make(chan ports.CursorChange, 1)
-	presented := make(chan ports.OutputPresented, 64)
-	// Outputs report on flips; relayPresented passes them to wayland and
-	// tells core about each flip (frames) to move running slides.
-	flips := make(chan ports.OutputPresented, 64)
-	frames := make(chan ports.OutputFrame, 8)
-	captures := make(chan ports.CaptureRequest, ports.MaxCaptureInflight)
-	// Every admitted capture owns one reply slot until Wayland consumes it.
-	// Keep this capacity tied to admission so output routing never waits.
-	captured := make(chan ports.CaptureDone, ports.MaxCaptureInflight)
-	outputFormats := make(chan ports.OutputFormats, 8)
-	outputHeads := make(chan ports.OutputHeads, 8)
-	leaseRequests := make(chan ports.LeaseMessage, 32)
-	leaseEvents := make(chan ports.LeaseMessage, 32)
-	applyOutput := make(chan ports.OutputApply, 8)
-	appliedOutput := make(chan ports.OutputApplied, 8)
-	idleInhibited := make(chan bool, 1)
-	idleActivity := make(chan struct{}, 1)
-	var scales chan ports.ScaleChanged
-	if hw != nil {
-		// Only real sessions save scales: headless runs never touch the config file.
-		scales = make(chan ports.ScaleChanged, 8)
-	}
-	ch := core.Channels{Security: security, Scales: scales, Client: client, Input: input, Output: output, Config: configChanges, Commands: commands, Spawn: spawn, Scenes: scenes, Layouts: layouts, Constraints: constraints, State: states, Workspaces: workspaces, ConfigErrors: configErrors, Terminal: !opts.NoTerminal, Clock: clock.System{}, Frames: frames}
 	c, err := core.New(opts.Config, ch)
 	if err != nil {
 		return err
 	} else {
 		captureAllow = captureallow.NewStore(captureallow.DefaultPath, captureAllowLog)
 	}
-	server, err := wayland.New(wayland.Options{CaptureAllow: captureAllow, WorkspaceIDs: wsIDs, Security: security, RuntimeDir: runtimeDir, DMABuf: dmabuf, SyncobjNode: renderNode(dmabuf.Device), Keymap: keymap, RepeatRate: opts.Config.Keyboard.RepeatRate, RepeatDelay: opts.Config.Keyboard.RepeatDelay}, wayland.Channels{SecurityChanges: securityChanges, SecurityEvents: securityEvents, Events: client, Commands: commands, Workspaces: workspaces, Contents: contents, Cursors: cursorChanges, Presented: presented, Captures: captures, Captured: captured, OutputFormats: outputFormats, OutputHeads: outputHeads, LeaseRequests: leaseRequests, LeaseEvents: leaseEvents, OutputApply: applyOutput, OutputApplied: appliedOutput, IdleInhibited: idleInhibited, IdleActivity: idleActivity}, logging.For(ctx, "wayland"))
+	server, err := wayland.New(wayland.Options{
+		CaptureAllow: captureAllow,
+		WorkspaceIDs: wsIDs,
+		Security:     security,
+		RuntimeDir:   runtimeDir,
+		DMABuf:       dmabuf,
+		SyncobjNode:  renderNode(dmabuf.Device),
+		Keymap:       keymap,
+		RepeatRate:   opts.Config.Keyboard.RepeatRate,
+		RepeatDelay:  opts.Config.Keyboard.RepeatDelay,
+	}, p.wayland(), logging.For(ctx, "wayland"))
 	if err != nil {
 		km.Close()
 		return err
 	child := launcher.New(childEnv, logging.For(ctx, "launcher"))
 	child.Security = security
 	if inject != nil {
-		inject(input)
+		inject(p.input)
 	}
 	var workers sync.WaitGroup
 	workers.Add(8)
 	if path == "" {
 		path = config.DefaultPath()
 	}
-	watched := make(chan ports.ConfigChanged, 8)
-	keymaps := make(chan *xkb.Keymap, 1)
-	deviceConfigs := make(chan ports.InputDevicesConfig, 1)
 	go func() {
 		defer workers.Done()
-		done <- config.Watch(ctx, path, watched, logging.For(ctx, "config"))
+		done <- config.Watch(ctx, path, p.watched, logging.For(ctx, "config"))
 	}()
-	if scales != nil {
+	if p.scales != nil {
 		workers.Add(1)
 		go func() {
 			defer workers.Done()
-			core.PersistScales(ctx, scales, config.ScaleStore{Path: path}, launcher.NewNotifier(ctx, childEnv, logging.For(ctx, "launcher")), clock.System{}, time.Second)
+			core.PersistScales(ctx, p.scales, config.ScaleStore{Path: path}, launcher.NewNotifier(ctx, childEnv, logging.For(ctx, "launcher")), clock.System{}, time.Second)
 		}()
 	}
-	filtered := make(chan ports.ConfigChanged, 8)
 	curs := newCursors(clock.System{}, opts.Config.Cursor.HideAfter)
 	defer curs.stop()
 	workers.Add(1)
 	go func() {
 		defer workers.Done()
-		relayConfig(ctx, opts.Config, watched, filtered, keymaps, deviceConfigs, curs, commands, logging.For(ctx, "config"))
+		relayConfig(ctx, opts.Config, p.watched, p.filtered, p.keymaps, p.deviceConfigs, curs, p.commands, logging.For(ctx, "config"))
 	}()
 	script := make(chan string)
 	go func() {
 			// Core sends the first layout once DRM reports the outputs.
 			var layout ports.Layout
 			select {
-			case layout = <-layouts:
+			case layout = <-p.layouts:
 			case <-ctx.Done():
 				km.Close()
 				return
 						log.Warn().Str("component", "sched").Err(err).Msg("input scheduling")
 					}
 				}
-				return libinput.Run(ctx, libinput.Options{Security: security, Seat: hw.seat, SeatName: hw.seat.Name(), Keymap: km, Keymaps: keymaps, Layout: layout, Layouts: layouts, Constraints: constraints, DeviceConfig: opts.Config.InputDevicesConfig, DeviceConfigs: deviceConfigs, Active: hw.seat.Subscribe(), MoveCursor: curs.move, Log: logging.For(ctx, "input"), LogMotion: logging.Enabled(ctx, "input-motion"), LogKeys: logging.Enabled(ctx, "input-keys")}, input)
+				return libinput.Run(ctx, libinput.Options{
+					Security:      security,
+					Seat:          hw.seat,
+					SeatName:      hw.seat.Name(),
+					Keymap:        km,
+					Keymaps:       p.keymaps,
+					Layout:        layout,
+					Layouts:       p.layouts,
+					Constraints:   p.constraints,
+					DeviceConfig:  opts.Config.InputDevicesConfig,
+					DeviceConfigs: p.deviceConfigs,
+					Active:        hw.seat.Subscribe(),
+					MoveCursor:    curs.move,
+					Log:           logging.For(ctx, "input"),
+					LogMotion:     logging.Enabled(ctx, "input-motion"),
+					LogKeys:       logging.Enabled(ctx, "input-keys"),
+				}, p.input)
 			})
 			return
 		}
-		if err := headlessinput.RunSecure(ctx, km, keymaps, script, input, layouts, func(o string, x, y float64) { curs.move(o, x, y, true) }, logging.For(ctx, "input"), security); err != nil && !errors.Is(err, context.Canceled) {
+		if err := headlessinput.RunSecure(ctx, km, p.keymaps, script, p.input, p.layouts, func(o string, x, y float64) { curs.move(o, x, y, true) }, logging.For(ctx, "input"), security); err != nil && !errors.Is(err, context.Canceled) {
 			select {
 			case done <- err:
 			case <-ctx.Done():
 			log.Warn().Err(err).Msg("capture allowlist not watched, restart to apply changes")
 		}
 	}()
-	go func() { defer workers.Done(); done <- child.Run(ctx, spawn) }()
+	go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
 	go func() { defer workers.Done(); done <- c.Run(ctx) }()
 	if xdisplay != nil {
 		workers.Add(1)
 		workers.Add(1)
 		go func() {
 			defer workers.Done()
-			if err := statefile.Run(ctx, statePath, states, wsIDs, logging.For(ctx, "statefile")); err != nil {
+			if err := statefile.Run(ctx, statePath, p.states, wsIDs, logging.For(ctx, "statefile")); err != nil {
 				// Scripts lose their state; the session goes on.
 				log.Warn().Err(err).Msg("state file disabled")
 			}
 		}()
 	}
-	go func() { defer workers.Done(); consumeScenes(ctx, scenes, configErrors, opts.testScenes, renderScenes) }()
+	go func() {
+		defer workers.Done()
+		consumeScenes(ctx, p.scenes, p.configErrors, opts.testScenes, p.renderScenes)
+	}()
 	workers.Add(1)
-	go func() { defer workers.Done(); relayPresented(ctx, flips, presented, frames) }()
+	go func() { defer workers.Done(); relayPresented(ctx, p.flips, p.presented, p.frames) }()
 	go func() {
 		defer workers.Done()
-		outputIO := outputChannels{security: security, securityChanges: securityChanges, securityEvents: securityEvents, events: output, scenes: renderScenes, contents: contents, cursorChanges: cursorChanges, presented: flips, captures: captures, captured: captured, formats: outputFormats, heads: outputHeads, leaseRequests: leaseRequests, leaseEvents: leaseEvents, reloads: filtered, configured: configChanges, requests: applyOutput, replies: appliedOutput}
+		outputIO := p.outputs(security)
 		apply := newOutputApply(newOutputOverrides(opts.Config, hw == nil), logging.For(ctx, "app"))
 		renderLog := logging.For(ctx, "render")
 		newRenderer := func(w, h int) (ports.Renderer, error) {
 		workers.Add(1)
 		go func() {
 			defer workers.Done()
-			reports := screensaver.Reports{Held: idleInhibited, Activity: idleActivity}
+			reports := screensaver.Reports{Held: p.idleInhibited, Activity: p.idleActivity}
 			screensaver.Serve(ctx, "", reports, screensaverRetry, logging.For(ctx, "screensaver"))
 		}()
 	}
 	cancel()
 	workers.Wait()
 	// The wayland server has stopped: no request can be queued any more.
-	drainCaptures(ctx, captures, nil)
+	drainCaptures(ctx, p.captures, nil)
 	// A keymap the input goroutine never took is still ours to free.
 	select {
-	case km := <-keymaps:
+	case km := <-p.keymaps:
 		km.Close()
 	default:
 	}

A internal/app/wiring.go +188 -0

@@ -0,0 +1,188 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"github.com/bnema/neferwl/internal/adapters/wayland"
	5	+	"github.com/bnema/neferwl/internal/adapters/xkb"
	6	+	"github.com/bnema/neferwl/internal/core"
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// pipes are the channels between a session's goroutines. newPipes makes
	11	+	// them all, with their buffer sizes; core, wayland and outputs hand each
	12	+	// side the ends it uses.
	13	+	type pipes struct {
	14	+	securityChanges chan ports.SecurityState
	15	+	securityEvents chan ports.SecurityBackendEvent
	16	+
	17	+	client chan ports.ClientEvent
	18	+	input chan ports.InputEvent
	19	+	output chan ports.OutputEvent
	20	+	commands chan ports.ClientCommand
	21	+	spawn chan ports.SpawnRequest
	22	+
	23	+	// Capacity 1: each holds the latest value its sender published.
	24	+	scenes chan []ports.Scene
	25	+	renderScenes chan []ports.Scene
	26	+	layouts chan ports.Layout
	27	+	constraints chan ports.PointerConstraint
	28	+	states chan ports.State
	29	+	workspaces chan ports.Workspaces
	30	+	cursorChanges chan ports.CursorChange
	31	+	keymaps chan *xkb.Keymap
	32	+	deviceConfigs chan ports.InputDevicesConfig
	33	+
	34	+	// Configuration: watched from the file, filtered by relayConfig, then
	35	+	// configured (overrides included) by outputs for core.
	36	+	watched chan ports.ConfigChanged
	37	+	filtered chan ports.ConfigChanged
	38	+	configured chan ports.ConfigChanged
	39	+	configErrors chan error
	40	+	// scales is nil outside a real session: headless runs never touch the
	41	+	// config file.
	42	+	scales chan ports.ScaleChanged
	43	+
	44	+	contents chan ports.SurfaceContent
	45	+	presented chan ports.OutputPresented
	46	+	// Outputs report on flips; relayPresented passes them to wayland
	47	+	// (presented) and tells core about each flip (frames) to move running
	48	+	// slides.
	49	+	flips chan ports.OutputPresented
	50	+	frames chan ports.OutputFrame
	51	+
	52	+	captures chan ports.CaptureRequest
	53	+	// Every admitted capture owns one reply slot until wayland consumes it.
	54	+	// Keep this capacity tied to admission so output routing never waits.
	55	+	captured chan ports.CaptureDone
	56	+
	57	+	outputFormats chan ports.OutputFormats
	58	+	outputHeads chan ports.OutputHeads
	59	+	leaseRequests chan ports.LeaseMessage
	60	+	leaseEvents chan ports.LeaseMessage
	61	+	applyOutput chan ports.OutputApply
	62	+	appliedOutput chan ports.OutputApplied
	63	+
	64	+	// D-Bus idle inhibition and SimulateUserActivity (screensaver), for
	65	+	// wayland's idle notifications. Activity sends never block: capacity 1
	66	+	// coalesces them.
	67	+	idleInhibited chan bool
	68	+	idleActivity chan struct{}
	69	+	}
	70	+
	71	+	func newPipes(realSession bool) *pipes {
	72	+	p := &pipes{
	73	+	securityChanges: make(chan ports.SecurityState, 8),
	74	+	securityEvents: make(chan ports.SecurityBackendEvent, 64),
	75	+
	76	+	client: make(chan ports.ClientEvent, 32),
	77	+	input: make(chan ports.InputEvent, 32),
	78	+	output: make(chan ports.OutputEvent, 32),
	79	+	commands: make(chan ports.ClientCommand, 32),
	80	+	spawn: make(chan ports.SpawnRequest, 32),
	81	+
	82	+	scenes: make(chan []ports.Scene, 1),
	83	+	renderScenes: make(chan []ports.Scene, 1),
	84	+	layouts: make(chan ports.Layout, 1),
	85	+	constraints: make(chan ports.PointerConstraint, 1),
	86	+	states: make(chan ports.State, 1),
	87	+	workspaces: make(chan ports.Workspaces, 1),
	88	+	cursorChanges: make(chan ports.CursorChange, 1),
	89	+	keymaps: make(chan *xkb.Keymap, 1),
	90	+	deviceConfigs: make(chan ports.InputDevicesConfig, 1),
	91	+
	92	+	watched: make(chan ports.ConfigChanged, 8),
	93	+	filtered: make(chan ports.ConfigChanged, 8),
	94	+	configured: make(chan ports.ConfigChanged, 8),
	95	+	configErrors: make(chan error, 8),
	96	+
	97	+	contents: make(chan ports.SurfaceContent, 64),
	98	+	presented: make(chan ports.OutputPresented, 64),
	99	+	flips: make(chan ports.OutputPresented, 64),
	100	+	frames: make(chan ports.OutputFrame, 8),
	101	+
	102	+	captures: make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
	103	+	captured: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
	104	+
	105	+	outputFormats: make(chan ports.OutputFormats, 8),
	106	+	outputHeads: make(chan ports.OutputHeads, 8),
	107	+	leaseRequests: make(chan ports.LeaseMessage, 32),
	108	+	leaseEvents: make(chan ports.LeaseMessage, 32),
	109	+	applyOutput: make(chan ports.OutputApply, 8),
	110	+	appliedOutput: make(chan ports.OutputApplied, 8),
	111	+
	112	+	idleInhibited: make(chan bool, 1),
	113	+	idleActivity: make(chan struct{}, 1),
	114	+	}
	115	+	if realSession {
	116	+	p.scales = make(chan ports.ScaleChanged, 8)
	117	+	}
	118	+	return p
	119	+	}
	120	+
	121	+	// core is core's side. Security, Terminal and Clock are not channels: the
	122	+	// caller sets them.
	123	+	func (p *pipes) core() core.Channels {
	124	+	return core.Channels{
	125	+	Client: p.client,
	126	+	Input: p.input,
	127	+	Output: p.output,
	128	+	Config: p.configured,
	129	+	Commands: p.commands,
	130	+	Spawn: p.spawn,
	131	+	Scenes: p.scenes,
	132	+	Layouts: p.layouts,
	133	+	Constraints: p.constraints,
	134	+	State: p.states,
	135	+	Workspaces: p.workspaces,
	136	+	ConfigErrors: p.configErrors,
	137	+	Scales: p.scales,
	138	+	Frames: p.frames,
	139	+	}
	140	+	}
	141	+
	142	+	// wayland is the wayland server's side.
	143	+	func (p *pipes) wayland() wayland.Channels {
	144	+	return wayland.Channels{
	145	+	SecurityChanges: p.securityChanges,
	146	+	SecurityEvents: p.securityEvents,
	147	+	Events: p.client,
	148	+	Commands: p.commands,
	149	+	Workspaces: p.workspaces,
	150	+	Contents: p.contents,
	151	+	Cursors: p.cursorChanges,
	152	+	Presented: p.presented,
	153	+	Captures: p.captures,
	154	+	Captured: p.captured,
	155	+	OutputFormats: p.outputFormats,
	156	+	OutputHeads: p.outputHeads,
	157	+	LeaseRequests: p.leaseRequests,
	158	+	LeaseEvents: p.leaseEvents,
	159	+	OutputApply: p.applyOutput,
	160	+	OutputApplied: p.appliedOutput,
	161	+	IdleInhibited: p.idleInhibited,
	162	+	IdleActivity: p.idleActivity,
	163	+	}
	164	+	}
	165	+
	166	+	// outputs is the outputs' side (DRM or headless).
	167	+	func (p *pipes) outputs(security ports.SessionSecurity) outputChannels {
	168	+	return outputChannels{
	169	+	security: security,
	170	+	securityChanges: p.securityChanges,
	171	+	securityEvents: p.securityEvents,
	172	+	events: p.output,
	173	+	scenes: p.renderScenes,
	174	+	contents: p.contents,
	175	+	cursorChanges: p.cursorChanges,
	176	+	presented: p.flips,
	177	+	captures: p.captures,
	178	+	captured: p.captured,
	179	+	formats: p.outputFormats,
	180	+	heads: p.outputHeads,
	181	+	leaseRequests: p.leaseRequests,
	182	+	leaseEvents: p.leaseEvents,
	183	+	reloads: p.filtered,
	184	+	configured: p.configured,
	185	+	requests: p.applyOutput,
	186	+	replies: p.appliedOutput,
	187	+	}
	188	+	}

@@ -0,0 +1,188 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"github.com/bnema/neferwl/internal/adapters/wayland"
	5	+	"github.com/bnema/neferwl/internal/adapters/xkb"
	6	+	"github.com/bnema/neferwl/internal/core"
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// pipes are the channels between a session's goroutines. newPipes makes
	11	+	// them all, with their buffer sizes; core, wayland and outputs hand each
	12	+	// side the ends it uses.
	13	+	type pipes struct {
	14	+	securityChanges chan ports.SecurityState
	15	+	securityEvents chan ports.SecurityBackendEvent
	16	+
	17	+	client chan ports.ClientEvent
	18	+	input chan ports.InputEvent
	19	+	output chan ports.OutputEvent
	20	+	commands chan ports.ClientCommand
	21	+	spawn chan ports.SpawnRequest
	22	+
	23	+	// Capacity 1: each holds the latest value its sender published.
	24	+	scenes chan []ports.Scene
	25	+	renderScenes chan []ports.Scene
	26	+	layouts chan ports.Layout
	27	+	constraints chan ports.PointerConstraint
	28	+	states chan ports.State
	29	+	workspaces chan ports.Workspaces
	30	+	cursorChanges chan ports.CursorChange
	31	+	keymaps chan *xkb.Keymap
	32	+	deviceConfigs chan ports.InputDevicesConfig
	33	+
	34	+	// Configuration: watched from the file, filtered by relayConfig, then
	35	+	// configured (overrides included) by outputs for core.
	36	+	watched chan ports.ConfigChanged
	37	+	filtered chan ports.ConfigChanged
	38	+	configured chan ports.ConfigChanged
	39	+	configErrors chan error
	40	+	// scales is nil outside a real session: headless runs never touch the
	41	+	// config file.
	42	+	scales chan ports.ScaleChanged
	43	+
	44	+	contents chan ports.SurfaceContent
	45	+	presented chan ports.OutputPresented
	46	+	// Outputs report on flips; relayPresented passes them to wayland
	47	+	// (presented) and tells core about each flip (frames) to move running
	48	+	// slides.
	49	+	flips chan ports.OutputPresented
	50	+	frames chan ports.OutputFrame
	51	+
	52	+	captures chan ports.CaptureRequest
	53	+	// Every admitted capture owns one reply slot until wayland consumes it.
	54	+	// Keep this capacity tied to admission so output routing never waits.
	55	+	captured chan ports.CaptureDone
	56	+
	57	+	outputFormats chan ports.OutputFormats
	58	+	outputHeads chan ports.OutputHeads
	59	+	leaseRequests chan ports.LeaseMessage
	60	+	leaseEvents chan ports.LeaseMessage
	61	+	applyOutput chan ports.OutputApply
	62	+	appliedOutput chan ports.OutputApplied
	63	+
	64	+	// D-Bus idle inhibition and SimulateUserActivity (screensaver), for
	65	+	// wayland's idle notifications. Activity sends never block: capacity 1
	66	+	// coalesces them.
	67	+	idleInhibited chan bool
	68	+	idleActivity chan struct{}
	69	+	}
	70	+
	71	+	func newPipes(realSession bool) *pipes {
	72	+	p := &pipes{
	73	+	securityChanges: make(chan ports.SecurityState, 8),
	74	+	securityEvents: make(chan ports.SecurityBackendEvent, 64),
	75	+
	76	+	client: make(chan ports.ClientEvent, 32),
	77	+	input: make(chan ports.InputEvent, 32),
	78	+	output: make(chan ports.OutputEvent, 32),
	79	+	commands: make(chan ports.ClientCommand, 32),
	80	+	spawn: make(chan ports.SpawnRequest, 32),
	81	+
	82	+	scenes: make(chan []ports.Scene, 1),
	83	+	renderScenes: make(chan []ports.Scene, 1),
	84	+	layouts: make(chan ports.Layout, 1),
	85	+	constraints: make(chan ports.PointerConstraint, 1),
	86	+	states: make(chan ports.State, 1),
	87	+	workspaces: make(chan ports.Workspaces, 1),
	88	+	cursorChanges: make(chan ports.CursorChange, 1),
	89	+	keymaps: make(chan *xkb.Keymap, 1),
	90	+	deviceConfigs: make(chan ports.InputDevicesConfig, 1),
	91	+
	92	+	watched: make(chan ports.ConfigChanged, 8),
	93	+	filtered: make(chan ports.ConfigChanged, 8),
	94	+	configured: make(chan ports.ConfigChanged, 8),
	95	+	configErrors: make(chan error, 8),
	96	+
	97	+	contents: make(chan ports.SurfaceContent, 64),
	98	+	presented: make(chan ports.OutputPresented, 64),
	99	+	flips: make(chan ports.OutputPresented, 64),
	100	+	frames: make(chan ports.OutputFrame, 8),
	101	+
	102	+	captures: make(chan ports.CaptureRequest, ports.MaxCaptureInflight),
	103	+	captured: make(chan ports.CaptureDone, ports.MaxCaptureInflight),
	104	+
	105	+	outputFormats: make(chan ports.OutputFormats, 8),
	106	+	outputHeads: make(chan ports.OutputHeads, 8),
	107	+	leaseRequests: make(chan ports.LeaseMessage, 32),
	108	+	leaseEvents: make(chan ports.LeaseMessage, 32),
	109	+	applyOutput: make(chan ports.OutputApply, 8),
	110	+	appliedOutput: make(chan ports.OutputApplied, 8),
	111	+
	112	+	idleInhibited: make(chan bool, 1),
	113	+	idleActivity: make(chan struct{}, 1),
	114	+	}
	115	+	if realSession {
	116	+	p.scales = make(chan ports.ScaleChanged, 8)
	117	+	}
	118	+	return p
	119	+	}
	120	+
	121	+	// core is core's side. Security, Terminal and Clock are not channels: the
	122	+	// caller sets them.
	123	+	func (p *pipes) core() core.Channels {
	124	+	return core.Channels{
	125	+	Client: p.client,
	126	+	Input: p.input,
	127	+	Output: p.output,
	128	+	Config: p.configured,
	129	+	Commands: p.commands,
	130	+	Spawn: p.spawn,
	131	+	Scenes: p.scenes,
	132	+	Layouts: p.layouts,
	133	+	Constraints: p.constraints,
	134	+	State: p.states,
	135	+	Workspaces: p.workspaces,
	136	+	ConfigErrors: p.configErrors,
	137	+	Scales: p.scales,
	138	+	Frames: p.frames,
	139	+	}
	140	+	}
	141	+
	142	+	// wayland is the wayland server's side.
	143	+	func (p *pipes) wayland() wayland.Channels {
	144	+	return wayland.Channels{
	145	+	SecurityChanges: p.securityChanges,
	146	+	SecurityEvents: p.securityEvents,
	147	+	Events: p.client,
	148	+	Commands: p.commands,
	149	+	Workspaces: p.workspaces,
	150	+	Contents: p.contents,
	151	+	Cursors: p.cursorChanges,
	152	+	Presented: p.presented,
	153	+	Captures: p.captures,
	154	+	Captured: p.captured,
	155	+	OutputFormats: p.outputFormats,
	156	+	OutputHeads: p.outputHeads,
	157	+	LeaseRequests: p.leaseRequests,
	158	+	LeaseEvents: p.leaseEvents,
	159	+	OutputApply: p.applyOutput,
	160	+	OutputApplied: p.appliedOutput,
	161	+	IdleInhibited: p.idleInhibited,
	162	+	IdleActivity: p.idleActivity,
	163	+	}
	164	+	}
	165	+
	166	+	// outputs is the outputs' side (DRM or headless).
	167	+	func (p *pipes) outputs(security ports.SessionSecurity) outputChannels {
	168	+	return outputChannels{
	169	+	security: security,
	170	+	securityChanges: p.securityChanges,
	171	+	securityEvents: p.securityEvents,
	172	+	events: p.output,
	173	+	scenes: p.renderScenes,
	174	+	contents: p.contents,
	175	+	cursorChanges: p.cursorChanges,
	176	+	presented: p.flips,
	177	+	captures: p.captures,
	178	+	captured: p.captured,
	179	+	formats: p.outputFormats,
	180	+	heads: p.outputHeads,
	181	+	leaseRequests: p.leaseRequests,
	182	+	leaseEvents: p.leaseEvents,
	183	+	reloads: p.filtered,
	184	+	configured: p.configured,
	185	+	requests: p.applyOutput,
	186	+	replies: p.appliedOutput,
	187	+	}
	188	+	}