Krolyxon
Commit feb7482

Branches

Refs →

budctlcratesbudctl-coresrcworker.rs

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