Krolyxon
Commit 6d9ef16

Branches

Refs →

budctlcratesbudctl-coresrcworker.rs

448 lines · 16.5 KB · Rust

Raw Blame History Download
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
9use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
10use std::thread;
11use std::time::{Duration, Instant};
12
13use crate::debug;
14use crate::devices::Session;
15use crate::error::{Error, Kind, Result};
16use 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.
21pub const IDLE: Duration = Duration::from_secs(10);
22/// How often an open session answers the buds while it waits for a command.
23const SERVE_EVERY: Duration = Duration::from_millis(100);
24
25#[derive(Debug, Clone)]
26pub 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)]
58pub 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)]
90pub 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.
104pub type Connect = Box<dyn FnMut(&dyn Fn(String)) -> Result<Box<dyn Session>> + Send>;
105
106/// The front end's side of the worker.
107pub 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.
118pub 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
127struct 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
139impl 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)]
303mod 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}