Commit 6d9ef16

budctl6d9ef16

Move the session worker into budctl-core

The thread that owns the session was budctl-gui's: it answers each
command with one event, answers the buds while it waits, and closes the
session after the idle time. The Android app needs the same one, so it
moves to budctl_core::worker, unchanged.

- worker::start spawns it, and a Report trait takes its events and log
  lines. Each front end brings its own Connect; the desktop app's goes
  through BlueZ.
- budctl-gui's worker.rs keeps the iced subscription: Message::Ready
  hands the app the worker's sender, then each event comes as
  Message::Worker.
- The worker's tests move with it, on std channels.

Author: krolyxon
Date: 2026-11-01 22:05 +0530
Parent: 8ec1f82
Commit: 6d9ef16b377c79c678e72e848ad3860a06951efc

+497 −451

5 files changed

Browse files Patch

−Context 3+

M CLAUDE.md +2 −2
M crates/budctl-core/src/lib.rs +1 −0
A crates/budctl-core/src/worker.rs +448 −0
M crates/budctl-gui/src/main.rs +11 −9
M crates/budctl-gui/src/worker.rs +35 −440

M CLAUDE.md View

M crates/budctl-core/src/lib.rs View

A crates/budctl-core/src/worker.rs View

M crates/budctl-gui/src/main.rs View

M crates/budctl-gui/src/worker.rs View

		@@ -33,7 +33,7 @@ The first supported device is the realme Buds T310. More devices are added a
33	33		## Stack
34	34
35	35		- Rust (edition 2024) in a cargo workspace:
36		−	- `crates/budctl-core`: library with transport, BlueZ and device plugins. It builds for Android too (`aarch64-linux-android`): BlueZ, the sockets and SDP are Linux-only (`cfg(target_os = "linux")`).
	36	+	- `crates/budctl-core`: library with transport, BlueZ, device plugins and the session worker. It builds for Android too (`aarch64-linux-android`): BlueZ, the sockets and SDP are Linux-only (`cfg(target_os = "linux")`).
37	37		- `crates/budctl-cli`: package and binary `budctl`.
38	38		- `crates/budctl-gui`: package and binary `budctl-gui`, built with iced 0.14 for Wayland only, with the CPU renderer (tiny-skia).
39	39		- On Linux, Bluetooth goes through BlueZ:
		@@ -123,7 +123,7 @@ Exit codes:
123	123		- `h`/`l` pick the previous or next choice, or turn a switch off or on, and `l` opens a row's page. Enter and Space toggle, step round the choices, or open.
124	124		- `J`/`K`, `Ctrl+Tab`, `gt`/`gT` and `1`–`5` change pages. `Esc` hides the cursor, `q` quits.
125	125		- The cursor shows only after a key press, and the first key only shows it, so nothing changes where it cannot be seen. A click hides it.
126		−	- Buds: `worker.rs` owns the session, on a thread of its own, and answers each command with one event, in order. It opens the session when a command needs it, answers the buds' own requests while it waits (`Session::serve`), and closes it 10 s after the last command (`gui.idle_seconds` in the config). A write shows "Saving…" until the buds' read-back arrives, and takes no second write meanwhile. A failure says why, with "Try again". `--demo` shows made-up buds; `--debug` hex-dumps frames like the CLI.
	126	+	- Buds: the worker (`budctl_core::worker`, run by `worker.rs` as an iced subscription) owns the session, on a thread of its own, and answers each command with one event, in order. It opens the session when a command needs it, answers the buds' own requests while it waits (`Session::serve`), and closes it 10 s after the last command (`gui.idle_seconds` in the config). A write shows "Saving…" until the buds' read-back arrives, and takes no second write meanwhile. A failure says why, with "Try again". `--demo` shows made-up buds; `--debug` hex-dumps frames like the CLI.
127	127		- Files: writes the built-in palette to `~/.config/budctl/theme.json` when missing (never overwrites it), and logs errors and session events to `~/.local/state/budctl/gui.log` (`log.rs`), replaced each launch. Frames stay on stderr with `--debug`. The CLI logs to stderr only, since status bars run it every few seconds.
128	128		- Speed: the CPU renderer repaints only what changed, one patch per changed thing, and it takes a canvas drawn afresh as changed on every frame. So every canvas keeps its drawing between frames (`kept.rs`), and cards sit on layers of their own (`ui::layer`), so a scroll paints a card once rather than once per row. Debug builds optimize the dependencies (`[profile.dev.package."*"]`); unoptimized, the window lags.
129	129		- Motion: Nothing's, per `motion.rs`: 150–250 ms, eased out on `cubic-bezier(0.25, 0.1, 0.25, 1)`, no bounce; things fade rather than slide. Small moves live in their widget (hovers and state changes on `motion::eased` surfaces; the switch, ring sweep, slider dots, blink and scan in `dots.rs`): a widget starts its tween on a redraw, asks for frames only while it runs, then the window rests. Fades of whole parts (page, dialog, error banner, open bands) are in `App`: `Look::fade`, with the `window::frames` subscription only while one runs. `gui.motion: false` turns it all off; the tests run without motion, so snapshots show settled states.

		@@ -16,6 +16,7 @@ mod error;
16	16		mod reader;
17	17		pub mod status;
18	18		pub mod transport;
	19	+	pub mod worker;
19	20
20	21		pub use error::{Error, Kind, Result};
21	22

		@@ -1,8 +1,8 @@
1	1		//! `budctl-gui`: control wireless earbuds from Linux, in a window.
2	2		//!
3		−	//! A thin layer over budctl-core, like the CLI: `worker.rs` talks to the buds on a thread of its
4		−	//! own, and the window shows what they report. The mouse comes first; vim-like keys work on top
5		−	//! (`keys.rs`). `--demo` shows made-up buds instead.
	3	+	//! A thin layer over budctl-core, like the CLI: core's worker talks to the buds on a thread of its
	4	+	//! own (`worker.rs` runs it), and the window shows what they report. The mouse comes first;
	5	+	//! vim-like keys work on top (`keys.rs`). `--demo` shows made-up buds instead.
6	6
7	7		mod buds;
8	8		mod dots;
		@@ -427,6 +427,8 @@ enum Message {
427	427		ShowHelp,
428	428		/// Read the buds again.
429	429		Refresh,
	430	+	/// The worker is up, and takes commands here.
	431	+	Ready(Sender<Command>),
430	432		Worker(Event),
431	433		/// The window got focus: the palette file may have changed meanwhile, and so may the buds.
432	434		Focused,
		@@ -636,6 +638,10 @@ impl App {
636	638		self.confirm_delete = None;
637	639		}
638	640		Message::Refresh => self.refresh(),
	641	+	Message::Ready(worker) => {
	642	+	self.worker = Some(worker);
	643	+	self.send(Command::Read, None);
	644	+	}
639	645		Message::Worker(event) => self.answer(event),
640	646		Message::Focused => {
641	647		self.refresh_palette();
		@@ -885,10 +891,6 @@ impl App {
885	891		/// Takes in what the worker reports.
886	892		fn answer(&mut self, event: Event) {
887	893		match event {
888		−	Event::Ready(worker) => {
889		−	self.worker = Some(worker);
890		−	self.send(Command::Read, None);
891		−	}
892	894		Event::Found(name) => self.buds.name = name,
893	895		Event::Lost(message) => {
894	896		self.buds.connected = false;
		@@ -1449,7 +1451,7 @@ impl App {
1449	1451		fn subscription(&self) -> Subscription<Message> {
1450	1452		let mut subscriptions = vec![event::listen_with(\|event, _status, _window\| on_event(event))];
1451	1453		if self.link != Link::Demo {
1452		−	subscriptions.push(Subscription::run_with((self.device, self.idle), worker::run).map(Message::Worker));
	1454	+	subscriptions.push(Subscription::run_with((self.device, self.idle), worker::run));
1453	1455		}
1454	1456		// Every frame while something fades, and none once it is over: the window rests.
1455	1457		if self.fading() {
		@@ -1848,7 +1850,7 @@ mod tests {
1848	1850		fn live() -> (App, Receiver<Command>) {
1849	1851		let mut app = App::new(None, None, false);
1850	1852		let (worker, commands) = mpsc::channel();
1851		−	let _ = app.update(Message::Worker(Event::Ready(worker)));
	1853	+	let _ = app.update(Message::Ready(worker));
1852	1854		(app, commands)
1853	1855		}
1854	1856

	@@ -0,0 +1,448 @@
1	+	//! The connection to the buds, on a thread of its own. Sessions block, and a write with its
2	+	//! read-back can take seconds, so a front end never waits on one: it sends [`Command`]s and gets
3	+	//! [`Event`]s back, in order, one per command. The desktop app and the Android app both run it.
4	+	//!
5	+	//! The session opens when a command needs it and closes after an idle time without one, so the
6	+	//! control channel is free for other apps (CLAUDE.md, connection behavior). While it is open, the
7	+	//! buds' own requests are answered.
8	+
9	+	use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
10	+	use std::thread;
11	+	use std::time::{Duration, Instant};
12	+
13	+	use crate::debug;
14	+	use crate::devices::Session;
15	+	use crate::error::{Error, Kind, Result};
16	+	use crate::status::{
17	+	DualDevice, EqBand, EqPreset, EqPresets, Gesture, NoiseControl, NoiseMode, Status, Switch, TouchControls,
18	+	};
19	+
20	+	/// How long the session stays open after the last command, unless the front end says otherwise.
21	+	pub const IDLE: Duration = Duration::from_secs(10);
22	+	/// How often an open session answers the buds while it waits for a command.
23	+	const SERVE_EVERY: Duration = Duration::from_millis(100);
24	+
25	+	#[derive(Debug, Clone)]
26	+	pub enum Command {
27	+	/// Read everything, connecting first if needed.
28	+	Read,
29	+	Noise(NoiseMode),
30	+	/// A switch's key, and whether to turn it on.
31	+	Switch(&'static str, bool),
32	+	/// Select a built-in EQ preset, by name.
33	+	Eq(String),
34	+	/// Save a custom EQ preset (`custom-<id>`) with these bands changed, which also selects it.
35	+	/// With no bands, it only selects it.
36	+	CustomEq(String, Vec<EqBand>),
37	+	/// Give a bud's gesture an action, by name: `left`, `double-tap`, `play-pause`.
38	+	Touch {
39	+	side: String,
40	+	gesture: String,
41	+	action: String,
42	+	},
43	+	/// Set the noise-control modes a hold steps through.
44	+	HoldCycle(Vec<String>),
45	+	/// Start (`true`) or stop ringing the buds.
46	+	Find(bool),
47	+	/// Set the bass level.
48	+	Bass(i8),
49	+	/// Create a custom EQ preset with this name.
50	+	CreateEq(String),
51	+	/// Give a custom EQ preset (`custom-<id>`) a new name.
52	+	RenameEq(String, String),
53	+	/// Delete a custom EQ preset (`custom-<id>`).
54	+	DeleteEq(String),
55	+	}
56	+
57	+	#[derive(Debug, Clone)]
58	+	pub enum Event {
59	+	/// The buds a new session is for, by the name their owner gave them. Answers no command.
60	+	Found(String),
61	+	/// The answer to [`Command::Read`]. Boxed: it is much larger than the other events.
62	+	Read(Box<Reading>),
63	+	/// The answer to [`Command::Noise`]: what the buds report after the write.
64	+	Noise(NoiseControl),
65	+	/// The answer to [`Command::Switch`]: the switch, as read back.
66	+	Switch(Switch),
67	+	/// The answer to [`Command::Eq`]: the preset now in use.
68	+	Eq(String),
69	+	/// The answer to [`Command::CustomEq`]: the preset, as read back.
70	+	CustomEq(EqPreset),
71	+	/// The answer to [`Command::Touch`]: the whole gesture table, as read back.
72	+	Touch(Vec<Gesture>),
73	+	/// The answer to [`Command::HoldCycle`]: the modes, as read back.
74	+	HoldCycle(Vec<String>),
75	+	/// The answer to [`Command::Find`]: whether the buds were told to ring.
76	+	Ringing(bool),
77	+	/// The answer to [`Command::Bass`]: the level the buds took.
78	+	Bass(i8),
79	+	/// The answer to [`Command::CreateEq`] and [`Command::DeleteEq`]: the whole preset list after
80	+	/// it (ids can change).
81	+	Presets(EqPresets),
82	+	/// The answer to a command that failed. The session is closed; the next command opens a
83	+	/// new one.
84	+	Failed { kind: Kind, message: String },
85	+	/// The link dropped while the session waited. Answers no command.
86	+	Lost(String),
87	+	}
88	+
89	+	#[derive(Debug, Clone)]
90	+	pub struct Reading {
91	+	pub status: Status,
92	+	/// The EQ presets, when the buds can list them: they name the custom ones.
93	+	pub eq: Option<EqPresets>,
94	+	/// The touch controls, when the buds report them.
95	+	pub touch: Option<TouchControls>,
96	+	/// The levels the bass level takes, when the buds have one to set.
97	+	pub bass_range: Option<(i8, i8)>,
98	+	/// The devices dual connection keeps, when the buds list them.
99	+	pub dual: Option<Vec<DualDevice>>,
100	+	}
101	+
102	+	/// Finds the buds and opens a session, calling `found` with their name once it is known. Each
103	+	/// front end brings its own: BlueZ on Linux, the app's Kotlin side on Android.
104	+	pub type Connect = Box<dyn FnMut(&dyn Fn(String)) -> Result<Box<dyn Session>> + Send>;
105	+
106	+	/// The front end's side of the worker.
107	+	pub trait Report: Send {
108	+	/// An answer to a command, or [`Event::Found`] or [`Event::Lost`].
109	+	fn event(&self, event: Event);
110	+
111	+	/// A line for the front end's log: each command, sessions opening and closing, failures.
112	+	fn log(&self, line: &str);
113	+	}
114	+
115	+	/// Starts the worker on a thread of its own. It takes commands from the sender it returns, and
116	+	/// ends, closing any session, once that sender is dropped. `idle` is how long a session stays
117	+	/// open after the last command.
118	+	pub fn start(connect: Connect, report: Box<dyn Report>, idle: Duration) -> std::io::Result<Sender<Command>> {
119	+	let (commands, inbox) = mpsc::channel();
120	+	let worker = Worker::new(connect, report, idle);
121	+	thread::Builder::new()
122	+	.name("budctl-buds".to_owned())
123	+	.spawn(move \|\| worker.run(&inbox))?;
124	+	Ok(commands)
125	+	}
126	+
127	+	struct Worker {
128	+	connect: Connect,
129	+	session: Option<Box<dyn Session>>,
130	+	report: Box<dyn Report>,
131	+	last_command: Instant,
132	+	/// How long the session stays open after the last command.
133	+	idle: Duration,
134	+	/// The buds once left the dual-connection list unanswered. They are not asked again, so
135	+	/// a reading does not wait for that timeout every time.
136	+	dual_unanswered: bool,
137	+	}
138	+
139	+	impl Worker {
140	+	fn new(connect: Connect, report: Box<dyn Report>, idle: Duration) -> Self {
141	+	Worker {
142	+	connect,
143	+	session: None,
144	+	report,
145	+	last_command: Instant::now(),
146	+	idle,
147	+	dual_unanswered: false,
148	+	}
149	+	}
150	+
151	+	/// Runs until the front end drops its sender.
152	+	fn run(mut self, inbox: &Receiver<Command>) {
153	+	loop {
154	+	let next = if self.session.is_some() {
155	+	inbox.recv_timeout(SERVE_EVERY)
156	+	} else {
157	+	inbox.recv().map_err(\|_\| RecvTimeoutError::Disconnected)
158	+	};
159	+	match next {
160	+	Ok(command) => {
161	+	self.last_command = Instant::now();
162	+	self.handle(command);
163	+	}
164	+	Err(RecvTimeoutError::Timeout) => self.wait(),
165	+	Err(RecvTimeoutError::Disconnected) => return,
166	+	}
167	+	}
168	+	}
169	+
170	+	/// Between commands: answer the buds, and close the session once it has been idle long
171	+	/// enough.
172	+	fn wait(&mut self) {
173	+	let Some(session) = self.session.as_mut() else {
174	+	return;
175	+	};
176	+	if self.last_command.elapsed() >= self.idle {
177	+	self.session = None;
178	+	let message = format!("session closed after {} s without a command", self.idle.as_secs());
179	+	debug::note(\|\| message.clone());
180	+	self.report.log(&message);
181	+	} else if let Err(error) = session.serve() {
182	+	self.session = None;
183	+	self.report.log(&format!("link lost: {error}"));
184	+	self.report.event(Event::Lost(error.to_string()));
185	+	}
186	+	}
187	+
188	+	fn handle(&mut self, command: Command) {
189	+	self.report.log(&format!("{command:?}"));
190	+	let answer = match command {
191	+	Command::Read => self.read().map(\|reading\| Event::Read(Box::new(reading))),
192	+	Command::Noise(mode) => self
193	+	.session()
194	+	.and_then(\|session\| session.set_noise_control(mode))
195	+	.map(Event::Noise),
196	+	Command::Switch(key, on) => self
197	+	.session()
198	+	.and_then(\|session\| session.set_switch(key, on))
199	+	.map(Event::Switch),
200	+	Command::Eq(preset) => self
201	+	.session()
202	+	.and_then(\|session\| session.set_eq(&preset))
203	+	.map(Event::Eq),
204	+	Command::CustomEq(preset, bands) => self
205	+	.session()
206	+	.and_then(\|session\| session.set_custom_eq(&preset, &bands))
207	+	.map(Event::CustomEq),
208	+	Command::Touch { side, gesture, action } => self
209	+	.session()
210	+	.and_then(\|session\| session.set_touch(&side, &gesture, &action, false))
211	+	.map(\|change\| Event::Touch(change.gestures)),
212	+	Command::CreateEq(name) => self
213	+	.session()
214	+	.and_then(\|session\| {
215	+	session.create_custom_eq(&name)?;
216	+	session.eq_presets()
217	+	})
218	+	.map(Event::Presets),
219	+	Command::DeleteEq(preset) => self
220	+	.session()
221	+	.and_then(\|session\| {
222	+	session.delete_custom_eq(&preset)?;
223	+	session.eq_presets()
224	+	})
225	+	.map(Event::Presets),
226	+	Command::RenameEq(preset, name) => self
227	+	.session()
228	+	.and_then(\|session\| session.rename_custom_eq(&preset, &name))
229	+	.map(Event::CustomEq),
230	+	Command::Bass(level) => self
231	+	.session()
232	+	.and_then(\|session\| session.set_bass_level(level))
233	+	.map(\|()\| Event::Bass(level)),
234	+	Command::Find(ring) => self
235	+	.session()
236	+	.and_then(\|session\| session.find(ring))
237	+	.map(\|()\| Event::Ringing(ring)),
238	+	Command::HoldCycle(modes) => {
239	+	let modes: Vec<&str> = modes.iter().map(String::as_str).collect();
240	+	self.session()
241	+	.and_then(\|session\| session.set_hold_cycle(&modes))
242	+	.map(Event::HoldCycle)
243	+	}
244	+	};
245	+	match answer {
246	+	Ok(event) => {
247	+	self.report.log(" done");
248	+	self.report.event(event);
249	+	}
250	+	Err(error) => {
251	+	self.report.log(&format!(" failed ({:?}): {error}", error.kind()));
252	+	self.session = None;
253	+	self.report.event(Event::Failed {
254	+	kind: error.kind(),
255	+	message: error.to_string(),
256	+	});
257	+	}
258	+	}
259	+	}
260	+
261	+	fn read(&mut self) -> Result<Reading> {
262	+	let ask_dual = !self.dual_unanswered;
263	+	let session = self.session()?;
264	+	let status = session.status()?;
265	+	let eq = session.eq_presets().ok();
266	+	let touch = session.touch_controls().ok();
267	+	let bass_range = session.bass_range();
268	+	let dual = match ask_dual.then(\|\| session.dual_devices()) {
269	+	Some(Ok(devices)) => Some(devices),
270	+	Some(Err(Error::Timeout(_))) => {
271	+	self.dual_unanswered = true;
272	+	None
273	+	}
274	+	Some(Err(_)) \| None => None,
275	+	};
276	+	Ok(Reading {
277	+	status,
278	+	eq,
279	+	touch,
280	+	bass_range,
281	+	dual,
282	+	})
283	+	}
284	+
285	+	/// The open session, opening one first if there is none.
286	+	fn session(&mut self) -> Result<&mut Box<dyn Session>> {
287	+	if self.session.is_none() {
288	+	let report = &self.report;
289	+	let found = \|name: String\| {
290	+	report.log(&format!(" connecting to {name}"));
291	+	report.event(Event::Found(name));
292	+	};
293	+	self.session = Some((self.connect)(&found)?);
294	+	self.report.log(" session open");
295	+	}
296	+	self.session
297	+	.as_mut()
298	+	.ok_or_else(\|\| Error::NotConnected("the session closed".to_owned()))
299	+	}
300	+	}
301	+
302	+	#[cfg(test)]
303	+	mod tests {
304	+	use super::*;
305	+	use std::sync::Arc;
306	+	use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
307	+
308	+	/// What the fake sessions report back to the test.
309	+	#[derive(Clone, Default)]
310	+	struct Probe {
311	+	opened: Arc<AtomicUsize>,
312	+	served: Arc<AtomicUsize>,
313	+	closed: Arc<AtomicBool>,
314	+	/// Make `serve` fail, as when the buds drop the link.
315	+	drop_link: Arc<AtomicBool>,
316	+	/// How often the dual-connection list was asked for. The fake buds never answer it.
317	+	dual_asked: Arc<AtomicUsize>,
318	+	}
319	+
320	+	struct Fake(Probe);
321	+
322	+	impl Session for Fake {
323	+	fn status(&mut self) -> Result<Status> {
324	+	Ok(Status::default())
325	+	}
326	+
327	+	fn dual_devices(&mut self) -> Result<Vec<DualDevice>> {
328	+	self.0.dual_asked.fetch_add(1, Ordering::SeqCst);
329	+	Err(Error::Timeout("the dual-connection list".to_owned()))
330	+	}
331	+
332	+	fn serve(&mut self) -> Result<()> {
333	+	self.0.served.fetch_add(1, Ordering::SeqCst);
334	+	if self.0.drop_link.load(Ordering::SeqCst) {
335	+	return Err(Error::NotConnected("the buds closed the control link".to_owned()));
336	+	}
337	+	Ok(())
338	+	}
339	+	}
340	+
341	+	impl Drop for Fake {
342	+	fn drop(&mut self) {
343	+	self.0.closed.store(true, Ordering::SeqCst);
344	+	}
345	+	}
346	+
347	+	/// A front end that only collects the events.
348	+	struct Collect(Sender<Event>);
349	+
350	+	impl Report for Collect {
351	+	fn event(&self, event: Event) {
352	+	let _ = self.0.send(event);
353	+	}
354	+
355	+	fn log(&self, _line: &str) {}
356	+	}
357	+
358	+	/// A worker on its own thread with fake buds, closing idle sessions after 200 ms.
359	+	fn start(probe: &Probe) -> (Sender<Command>, Receiver<Event>, thread::JoinHandle<()>) {
360	+	let fake = probe.clone();
361	+	let connect: Connect = Box::new(move \|found\| {
362	+	found("Fake buds".to_owned());
363	+	fake.opened.fetch_add(1, Ordering::SeqCst);
364	+	fake.closed.store(false, Ordering::SeqCst);
365	+	Ok(Box::new(Fake(fake.clone())))
366	+	});
367	+	let (events, outbox) = mpsc::channel();
368	+	let (commands, inbox) = mpsc::channel();
369	+	let worker = Worker::new(connect, Box::new(Collect(events)), Duration::from_millis(200));
370	+	let thread = thread::spawn(move \|\| worker.run(&inbox));
371	+	(commands, outbox, thread)
372	+	}
373	+
374	+	/// The next event, waiting at most 5 s.
375	+	fn next(events: &Receiver<Event>) -> Option<Event> {
376	+	events.recv_timeout(Duration::from_secs(5)).ok()
377	+	}
378	+
379	+	/// Waits, at most 5 s, for `done`. A receive that times out stands in for a sleep.
380	+	fn wait_for(done: impl Fn() -> bool) {
381	+	let (_keep, never) = mpsc::channel::<()>();
382	+	let deadline = Instant::now() + Duration::from_secs(5);
383	+	while !done() {
384	+	assert!(Instant::now() < deadline, "gave up waiting");
385	+	let _ = never.recv_timeout(Duration::from_millis(10));
386	+	}
387	+	}
388	+
389	+	#[test]
390	+	fn an_idle_session_closes_and_the_next_command_opens_another() {
391	+	let probe = Probe::default();
392	+	let (commands, events, thread) = start(&probe);
393	+
394	+	commands.send(Command::Read).unwrap();
395	+	assert!(matches!(next(&events), Some(Event::Found(name)) if name == "Fake buds"));
396	+	assert!(matches!(next(&events), Some(Event::Read(_))));
397	+	assert_eq!(probe.opened.load(Ordering::SeqCst), 1);
398	+
399	+	// While it waits, it answers the buds; then it closes the session.
400	+	wait_for(\|\| probe.closed.load(Ordering::SeqCst));
401	+	assert!(probe.served.load(Ordering::SeqCst) > 0);
402	+
403	+	commands.send(Command::Read).unwrap();
404	+	assert!(matches!(next(&events), Some(Event::Found(_))));
405	+	assert!(matches!(next(&events), Some(Event::Read(_))));
406	+	assert_eq!(probe.opened.load(Ordering::SeqCst), 2);
407	+
408	+	// The front end goes away: the worker ends.
409	+	drop(commands);
410	+	thread.join().unwrap();
411	+	}
412	+
413	+	#[test]
414	+	fn a_dual_list_left_unanswered_is_not_asked_for_again() {
415	+	let probe = Probe::default();
416	+	let (commands, events, thread) = start(&probe);
417	+	commands.send(Command::Read).unwrap();
418	+	commands.send(Command::Read).unwrap();
419	+	assert!(matches!(next(&events), Some(Event::Found(_))));
420	+	for _ in 0..2 {
421	+	let Some(Event::Read(reading)) = next(&events) else {
422	+	panic!("a reading expected");
423	+	};
424	+	assert!(reading.dual.is_none());
425	+	}
426	+	assert_eq!(probe.dual_asked.load(Ordering::SeqCst), 1);
427	+
428	+	drop(commands);
429	+	thread.join().unwrap();
430	+	}
431	+
432	+	#[test]
433	+	fn a_dropped_link_is_reported_once() {
434	+	let probe = Probe::default();
435	+	let (commands, events, thread) = start(&probe);
436	+	commands.send(Command::Read).unwrap();
437	+	let _ = next(&events);
438	+	let _ = next(&events);
439	+
440	+	probe.drop_link.store(true, Ordering::SeqCst);
441	+	assert!(matches!(next(&events), Some(Event::Lost(_))));
442	+	assert!(probe.closed.load(Ordering::SeqCst));
443	+
444	+	drop(commands);
445	+	thread.join().unwrap();
446	+	assert!(events.recv().is_err(), "nothing more after the link dropped");
447	+	}
448	+	}

		@@ -1,140 +1,56 @@
1		−	//! The connection to the buds, on a thread of its own. budctl-core blocks, and a write with its
2		−	//! read-back can take seconds, so the window never waits on it: it sends [`Command`]s and gets
3		−	//! [`Event`]s back, in order, one per command.
4		−	//!
5		−	//! The session opens when a command needs it and closes after [`IDLE`] without one, so the
6		−	//! control channel is free for other apps (CLAUDE.md, connection behavior). While it is open, the
7		−	//! buds' own requests are answered.
	1	+	//! The connection to the buds: budctl-core's worker, which owns the session on a thread of its
	2	+	//! own, run as an iced subscription and connected through BlueZ.
8	3
9		−	use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
10		−	use std::thread;
11		−	use std::time::{Duration, Instant};
	4	+	use std::time::Duration;
12	5
13	6		use budctl_core::bt::BdAddr;
14		−	use budctl_core::devices::Session;
15		−	use budctl_core::status::{
16		−	DualDevice, EqBand, EqPreset, EqPresets, Gesture, NoiseControl, NoiseMode, Status, Switch, TouchControls,
17		−	};
18		−	use budctl_core::{Error, Kind, Result};
	7	+	use budctl_core::worker::{self, Connect, Report};
	8	+	pub use budctl_core::worker::{Command, Event, IDLE, Reading};
19	9		use iced::futures::channel::mpsc as channel;
20	10		use iced::futures::{SinkExt, Stream, StreamExt};
21	11
22		−	use crate::log;
	12	+	use crate::{Message, log};
23	13
24		−	/// How long the session stays open after the last command, unless the config says otherwise.
25		−	pub const IDLE: Duration = Duration::from_secs(10);
26		−	/// How often an open session answers the buds while it waits for a command.
27		−	const SERVE_EVERY: Duration = Duration::from_millis(100);
28		−
29		−	#[derive(Debug, Clone)]
30		−	pub enum Command {
31		−	/// Read everything, connecting first if needed.
32		−	Read,
33		−	Noise(NoiseMode),
34		−	/// A switch's key, and whether to turn it on.
35		−	Switch(&'static str, bool),
36		−	/// Select a built-in EQ preset, by name.
37		−	Eq(String),
38		−	/// Save a custom EQ preset (`custom-<id>`) with these bands changed, which also selects it.
39		−	/// With no bands, it only selects it.
40		−	CustomEq(String, Vec<EqBand>),
41		−	/// Give a bud's gesture an action, by name: `left`, `double-tap`, `play-pause`.
42		−	Touch {
43		−	side: String,
44		−	gesture: String,
45		−	action: String,
46		−	},
47		−	/// Set the noise-control modes a hold steps through.
48		−	HoldCycle(Vec<String>),
49		−	/// Start (`true`) or stop ringing the buds.
50		−	Find(bool),
51		−	/// Set the bass level.
52		−	Bass(i8),
53		−	/// Create a custom EQ preset with this name.
54		−	CreateEq(String),
55		−	/// Give a custom EQ preset (`custom-<id>`) a new name.
56		−	RenameEq(String, String),
57		−	/// Delete a custom EQ preset (`custom-<id>`).
58		−	DeleteEq(String),
59		−	}
60		−
61		−	#[derive(Debug, Clone)]
62		−	pub enum Event {
63		−	/// The worker takes commands here.
64		−	Ready(Sender<Command>),
65		−	/// The buds a new session is for, by the name their owner gave them. Answers no command.
66		−	Found(String),
67		−	/// The answer to [`Command::Read`]. Boxed: it is much larger than the other events.
68		−	Read(Box<Reading>),
69		−	/// The answer to [`Command::Noise`]: what the buds report after the write.
70		−	Noise(NoiseControl),
71		−	/// The answer to [`Command::Switch`]: the switch, as read back.
72		−	Switch(Switch),
73		−	/// The answer to [`Command::Eq`]: the preset now in use.
74		−	Eq(String),
75		−	/// The answer to [`Command::CustomEq`]: the preset, as read back.
76		−	CustomEq(EqPreset),
77		−	/// The answer to [`Command::Touch`]: the whole gesture table, as read back.
78		−	Touch(Vec<Gesture>),
79		−	/// The answer to [`Command::HoldCycle`]: the modes, as read back.
80		−	HoldCycle(Vec<String>),
81		−	/// The answer to [`Command::Find`]: whether the buds were told to ring.
82		−	Ringing(bool),
83		−	/// The answer to [`Command::Bass`]: the level the buds took.
84		−	Bass(i8),
85		−	/// The answer to [`Command::CreateEq`] and [`Command::DeleteEq`]: the whole preset list after
86		−	/// it (ids can change).
87		−	Presets(EqPresets),
88		−	/// The answer to a command that failed. The session is closed; the next command opens a
89		−	/// new one.
90		−	Failed { kind: Kind, message: String },
91		−	/// The link dropped while the session waited. Answers no command.
92		−	Lost(String),
93		−	}
94		−
95		−	#[derive(Debug, Clone)]
96		−	pub struct Reading {
97		−	pub status: Status,
98		−	/// The EQ presets, when the buds can list them: they name the custom ones.
99		−	pub eq: Option<EqPresets>,
100		−	/// The touch controls, when the buds report them.
101		−	pub touch: Option<TouchControls>,
102		−	/// The levels the bass level takes, when the buds have one to set.
103		−	pub bass_range: Option<(i8, i8)>,
104		−	/// The devices dual connection keeps, when the buds list them.
105		−	pub dual: Option<Vec<DualDevice>>,
106		−	}
107		−
108		−	/// The worker, as a stream of events for an iced subscription. `device` picks the buds, as
109		−	/// `budctl --device` does; `idle` is how long a session stays open after the last command.
110		−	pub fn run(&(device, idle): &(Option<BdAddr>, Duration)) -> impl Stream<Item = Event> + use<> {
111		−	iced::stream::channel(16, async move \|mut output: channel::Sender<Event>\| {
112		−	let (commands, inbox) = mpsc::channel();
	14	+	/// The worker, as a stream of messages for an iced subscription: [`Message::Ready`] once it takes
	15	+	/// commands, then each of its events. `device` picks the buds, as `budctl --device` does; `idle`
	16	+	/// is how long a session stays open after the last command.
	17	+	pub fn run(&(device, idle): &(Option<BdAddr>, Duration)) -> impl Stream<Item = Message> + use<> {
	18	+	iced::stream::channel(16, async move \|mut output: channel::Sender<Message>\| {
113	19		let (events, mut outbox) = channel::unbounded();
114		−	if output.send(Event::Ready(commands)).await.is_err() {
115		−	return;
116		−	}
117		−	let worker = Worker::new(bluetooth(device), events, idle);
118		−	let started = thread::Builder::new()
119		−	.name("budctl-buds".to_owned())
120		−	.spawn(move \|\| worker.run(&inbox));
121		−	if let Err(e) = started {
122		−	let message = format!("could not start the Bluetooth thread: {e}");
123		−	let _ = output.send(Event::Lost(message)).await;
	20	+	let commands = match worker::start(bluetooth(device), Box::new(Window(events)), idle) {
	21	+	Ok(commands) => commands,
	22	+	Err(e) => {
	23	+	let message = format!("could not start the Bluetooth thread: {e}");
	24	+	let _ = output.send(Message::Worker(Event::Lost(message))).await;
	25	+	return;
	26	+	}
	27	+	};
	28	+	if output.send(Message::Ready(commands)).await.is_err() {
124	29		return;
125	30		}
126	31		while let Some(event) = outbox.next().await {
127		−	if output.send(event).await.is_err() {
	32	+	if output.send(Message::Worker(event)).await.is_err() {
128	33		break;
129	34		}
130	35		}
131	36		})
132	37		}
133	38
134		−	/// Finds the buds and opens a session, calling `found` with their name once it is known.
135		−	type Connect = Box<dyn FnMut(&dyn Fn(String)) -> Result<Box<dyn Session>> + Send>;
	39	+	/// The window's side of the worker: events go to the subscription, lines to the GUI's log.
	40	+	struct Window(channel::UnboundedSender<Event>);
136	41
137		−	/// Over Bluetooth, through BlueZ: what the window uses. Tests use a fake.
	42	+	impl Report for Window {
	43	+	fn event(&self, event: Event) {
	44	+	// An error only means the window is gone; the worker ends once its sender is dropped.
	45	+	let _ = self.0.unbounded_send(event);
	46	+	}
	47	+
	48	+	fn log(&self, line: &str) {
	49	+	log::line(line);
	50	+	}
	51	+	}
	52	+
	53	+	/// Over Bluetooth, through BlueZ: what the window uses.
138	54		fn bluetooth(device: Option<BdAddr>) -> Connect {
139	55		Box::new(move \|found\| {
140	56		let (device, model) = budctl_core::select(budctl_core::scan()?, device)?;
		@@ -142,324 +58,3 @@ fn bluetooth(device: Option<BdAddr>) -> Connect {
142	58		budctl_core::connect(&device, model)
143	59		})
144	60		}
145		−
146		−	struct Worker {
147		−	connect: Connect,
148		−	session: Option<Box<dyn Session>>,
149		−	events: channel::UnboundedSender<Event>,
150		−	last_command: Instant,
151		−	/// How long the session stays open after the last command.
152		−	idle: Duration,
153		−	/// The buds once left the dual-connection list unanswered. They are not asked again, so
154		−	/// a reading does not wait for that timeout every time.
155		−	dual_unanswered: bool,
156		−	}
157		−
158		−	impl Worker {
159		−	fn new(connect: Connect, events: channel::UnboundedSender<Event>, idle: Duration) -> Self {
160		−	Worker {
161		−	connect,
162		−	session: None,
163		−	events,
164		−	last_command: Instant::now(),
165		−	idle,
166		−	dual_unanswered: false,
167		−	}
168		−	}
169		−
170		−	/// Runs until the window is gone.
171		−	fn run(mut self, inbox: &Receiver<Command>) {
172		−	loop {
173		−	let next = if self.session.is_some() {
174		−	inbox.recv_timeout(SERVE_EVERY)
175		−	} else {
176		−	inbox.recv().map_err(\|_\| RecvTimeoutError::Disconnected)
177		−	};
178		−	match next {
179		−	Ok(command) => {
180		−	self.last_command = Instant::now();
181		−	self.handle(command);
182		−	}
183		−	Err(RecvTimeoutError::Timeout) => self.wait(),
184		−	Err(RecvTimeoutError::Disconnected) => return,
185		−	}
186		−	}
187		−	}
188		−
189		−	/// Between commands: answer the buds, and close the session once it has been idle long
190		−	/// enough.
191		−	fn wait(&mut self) {
192		−	let Some(session) = self.session.as_mut() else {
193		−	return;
194		−	};
195		−	if self.last_command.elapsed() >= self.idle {
196		−	self.session = None;
197		−	let message = format!("session closed after {} s without a command", self.idle.as_secs());
198		−	if budctl_core::debug::enabled() {
199		−	eprintln!("# {message}");
200		−	}
201		−	log::line(message);
202		−	} else if let Err(error) = session.serve() {
203		−	self.session = None;
204		−	log::line(format!("link lost: {error}"));
205		−	self.send(Event::Lost(error.to_string()));
206		−	}
207		−	}
208		−
209		−	fn handle(&mut self, command: Command) {
210		−	log::line(format!("{command:?}"));
211		−	let answer = match command {
212		−	Command::Read => self.read().map(\|reading\| Event::Read(Box::new(reading))),
213		−	Command::Noise(mode) => self
214		−	.session()
215		−	.and_then(\|session\| session.set_noise_control(mode))
216		−	.map(Event::Noise),
217		−	Command::Switch(key, on) => self
218		−	.session()
219		−	.and_then(\|session\| session.set_switch(key, on))
220		−	.map(Event::Switch),
221		−	Command::Eq(preset) => self
222		−	.session()
223		−	.and_then(\|session\| session.set_eq(&preset))
224		−	.map(Event::Eq),
225		−	Command::CustomEq(preset, bands) => self
226		−	.session()
227		−	.and_then(\|session\| session.set_custom_eq(&preset, &bands))
228		−	.map(Event::CustomEq),
229		−	Command::Touch { side, gesture, action } => self
230		−	.session()
231		−	.and_then(\|session\| session.set_touch(&side, &gesture, &action, false))
232		−	.map(\|change\| Event::Touch(change.gestures)),
233		−	Command::CreateEq(name) => self
234		−	.session()
235		−	.and_then(\|session\| {
236		−	session.create_custom_eq(&name)?;
237		−	session.eq_presets()
238		−	})
239		−	.map(Event::Presets),
240		−	Command::DeleteEq(preset) => self
241		−	.session()
242		−	.and_then(\|session\| {
243		−	session.delete_custom_eq(&preset)?;
244		−	session.eq_presets()
245		−	})
246		−	.map(Event::Presets),
247		−	Command::RenameEq(preset, name) => self
248		−	.session()
249		−	.and_then(\|session\| session.rename_custom_eq(&preset, &name))
250		−	.map(Event::CustomEq),
251		−	Command::Bass(level) => self
252		−	.session()
253		−	.and_then(\|session\| session.set_bass_level(level))
254		−	.map(\|()\| Event::Bass(level)),
255		−	Command::Find(ring) => self
256		−	.session()
257		−	.and_then(\|session\| session.find(ring))
258		−	.map(\|()\| Event::Ringing(ring)),
259		−	Command::HoldCycle(modes) => {
260		−	let modes: Vec<&str> = modes.iter().map(String::as_str).collect();
261		−	self.session()
262		−	.and_then(\|session\| session.set_hold_cycle(&modes))
263		−	.map(Event::HoldCycle)
264		−	}
265		−	};
266		−	match answer {
267		−	Ok(event) => {
268		−	log::line(" done");
269		−	self.send(event);
270		−	}
271		−	Err(error) => {
272		−	log::line(format!(" failed ({:?}): {error}", error.kind()));
273		−	self.session = None;
274		−	self.send(Event::Failed {
275		−	kind: error.kind(),
276		−	message: error.to_string(),
277		−	});
278		−	}
279		−	}
280		−	}
281		−
282		−	fn read(&mut self) -> Result<Reading> {
283		−	let ask_dual = !self.dual_unanswered;
284		−	let session = self.session()?;
285		−	let status = session.status()?;
286		−	let eq = session.eq_presets().ok();
287		−	let touch = session.touch_controls().ok();
288		−	let bass_range = session.bass_range();
289		−	let dual = match ask_dual.then(\|\| session.dual_devices()) {
290		−	Some(Ok(devices)) => Some(devices),
291		−	Some(Err(Error::Timeout(_))) => {
292		−	self.dual_unanswered = true;
293		−	None
294		−	}
295		−	Some(Err(_)) \| None => None,
296		−	};
297		−	Ok(Reading {
298		−	status,
299		−	eq,
300		−	touch,
301		−	bass_range,
302		−	dual,
303		−	})
304		−	}
305		−
306		−	/// The open session, opening one first if there is none.
307		−	fn session(&mut self) -> Result<&mut Box<dyn Session>> {
308		−	if self.session.is_none() {
309		−	let events = self.events.clone();
310		−	let found = move \|name: String\| {
311		−	log::line(format!(" connecting to {name}"));
312		−	let _ = events.unbounded_send(Event::Found(name));
313		−	};
314		−	self.session = Some((self.connect)(&found)?);
315		−	log::line(" session open");
316		−	}
317		−	self.session
318		−	.as_mut()
319		−	.ok_or_else(\|\| Error::NotConnected("the session closed".to_owned()))
320		−	}
321		−
322		−	fn send(&self, event: Event) {
323		−	// An error only means the window is gone; `run` ends on its next receive.
324		−	let _ = self.events.unbounded_send(event);
325		−	}
326		−	}
327		−
328		−	#[cfg(test)]
329		−	mod tests {
330		−	use super::*;
331		−	use iced::futures::executor::block_on;
332		−	use std::sync::Arc;
333		−	use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
334		−
335		−	/// What the fake sessions report back to the test.
336		−	#[derive(Clone, Default)]
337		−	struct Probe {
338		−	opened: Arc<AtomicUsize>,
339		−	served: Arc<AtomicUsize>,
340		−	closed: Arc<AtomicBool>,
341		−	/// Make `serve` fail, as when the buds drop the link.
342		−	drop_link: Arc<AtomicBool>,
343		−	/// How often the dual-connection list was asked for. The fake buds never answer it.
344		−	dual_asked: Arc<AtomicUsize>,
345		−	}
346		−
347		−	struct Fake(Probe);
348		−
349		−	impl Session for Fake {
350		−	fn status(&mut self) -> Result<Status> {
351		−	Ok(Status::default())
352		−	}
353		−
354		−	fn dual_devices(&mut self) -> Result<Vec<DualDevice>> {
355		−	self.0.dual_asked.fetch_add(1, Ordering::SeqCst);
356		−	Err(Error::Timeout("the dual-connection list".to_owned()))
357		−	}
358		−
359		−	fn serve(&mut self) -> Result<()> {
360		−	self.0.served.fetch_add(1, Ordering::SeqCst);
361		−	if self.0.drop_link.load(Ordering::SeqCst) {
362		−	return Err(Error::NotConnected("the buds closed the control link".to_owned()));
363		−	}
364		−	Ok(())
365		−	}
366		−	}
367		−
368		−	impl Drop for Fake {
369		−	fn drop(&mut self) {
370		−	self.0.closed.store(true, Ordering::SeqCst);
371		−	}
372		−	}
373		−
374		−	/// A worker on its own thread with fake buds, closing idle sessions after 200 ms.
375		−	fn start(
376		−	probe: &Probe,
377		−	) -> (
378		−	Sender<Command>,
379		−	channel::UnboundedReceiver<Event>,
380		−	thread::JoinHandle<()>,
381		−	) {
382		−	let fake = probe.clone();
383		−	let connect: Connect = Box::new(move \|found\| {
384		−	found("Fake buds".to_owned());
385		−	fake.opened.fetch_add(1, Ordering::SeqCst);
386		−	fake.closed.store(false, Ordering::SeqCst);
387		−	Ok(Box::new(Fake(fake.clone())))
388		−	});
389		−	let (events, outbox) = channel::unbounded();
390		−	let (commands, inbox) = mpsc::channel();
391		−	let worker = Worker::new(connect, events, Duration::from_millis(200));
392		−	let thread = thread::spawn(move \|\| worker.run(&inbox));
393		−	(commands, outbox, thread)
394		−	}
395		−
396		−	/// Waits, at most 5 s, for `done`. A receive that times out stands in for a sleep.
397		−	fn wait_for(done: impl Fn() -> bool) {
398		−	let (_keep, never) = mpsc::channel::<()>();
399		−	let deadline = Instant::now() + Duration::from_secs(5);
400		−	while !done() {
401		−	assert!(Instant::now() < deadline, "gave up waiting");
402		−	let _ = never.recv_timeout(Duration::from_millis(10));
403		−	}
404		−	}
405		−
406		−	#[test]
407		−	fn an_idle_session_closes_and_the_next_command_opens_another() {
408		−	let probe = Probe::default();
409		−	let (commands, mut events, thread) = start(&probe);
410		−
411		−	commands.send(Command::Read).unwrap();
412		−	assert!(matches!(block_on(events.next()), Some(Event::Found(name)) if name == "Fake buds"));
413		−	assert!(matches!(block_on(events.next()), Some(Event::Read(_))));
414		−	assert_eq!(probe.opened.load(Ordering::SeqCst), 1);
415		−
416		−	// While it waits, it answers the buds; then it closes the session.
417		−	wait_for(\|\| probe.closed.load(Ordering::SeqCst));
418		−	assert!(probe.served.load(Ordering::SeqCst) > 0);
419		−
420		−	commands.send(Command::Read).unwrap();
421		−	assert!(matches!(block_on(events.next()), Some(Event::Found(_))));
422		−	assert!(matches!(block_on(events.next()), Some(Event::Read(_))));
423		−	assert_eq!(probe.opened.load(Ordering::SeqCst), 2);
424		−
425		−	// The window goes away: the worker ends.
426		−	drop(commands);
427		−	thread.join().unwrap();
428		−	}
429		−
430		−	#[test]
431		−	fn a_dual_list_left_unanswered_is_not_asked_for_again() {
432		−	let probe = Probe::default();
433		−	let (commands, mut events, thread) = start(&probe);
434		−	commands.send(Command::Read).unwrap();
435		−	commands.send(Command::Read).unwrap();
436		−	assert!(matches!(block_on(events.next()), Some(Event::Found(_))));
437		−	for _ in 0..2 {
438		−	let Some(Event::Read(reading)) = block_on(events.next()) else {
439		−	panic!("a reading expected");
440		−	};
441		−	assert!(reading.dual.is_none());
442		−	}
443		−	assert_eq!(probe.dual_asked.load(Ordering::SeqCst), 1);
444		−
445		−	drop(commands);
446		−	thread.join().unwrap();
447		−	}
448		−
449		−	#[test]
450		−	fn a_dropped_link_is_reported_once() {
451		−	let probe = Probe::default();
452		−	let (commands, mut events, thread) = start(&probe);
453		−	commands.send(Command::Read).unwrap();
454		−	let _ = block_on(events.next());
455		−	let _ = block_on(events.next());
456		−
457		−	probe.drop_link.store(true, Ordering::SeqCst);
458		−	assert!(matches!(block_on(events.next()), Some(Event::Lost(_))));
459		−	assert!(probe.closed.load(Ordering::SeqCst));
460		−
461		−	drop(commands);
462		−	thread.join().unwrap();
463		−	assert!(block_on(events.next()).is_none(), "nothing more after the link dropped");
464		−	}
465		−	}

budctl/6d9ef16

Move the session worker into budctl-core

budctl6d9ef16