Krolyxon
Commit f7d0d70

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

465 lines · 17.1 KB · Rust

Raw Blame History Download
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.
8
9use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
10use std::thread;
11use std::time::{Duration, Instant};
12
13use budctl_core::bt::BdAddr;
14use budctl_core::devices::Session;
15use budctl_core::status::{
16 DualDevice, EqBand, EqPreset, EqPresets, Gesture, NoiseControl, NoiseMode, Status, Switch, TouchControls,
17};
18use budctl_core::{Error, Kind, Result};
19use iced::futures::channel::mpsc as channel;
20use iced::futures::{SinkExt, Stream, StreamExt};
21
22use crate::log;
23
24/// How long the session stays open after the last command, unless the config says otherwise.
25pub const IDLE: Duration = Duration::from_secs(10);
26/// How often an open session answers the buds while it waits for a command.
27const SERVE_EVERY: Duration = Duration::from_millis(100);
28
29#[derive(Debug, Clone)]
30pub 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)]
62pub 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)]
96pub 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.
110pub 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();
113 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;
124 return;
125 }
126 while let Some(event) = outbox.next().await {
127 if output.send(event).await.is_err() {
128 break;
129 }
130 }
131 })
132}
133
134/// Finds the buds and opens a session, calling `found` with their name once it is known.
135type Connect = Box<dyn FnMut(&dyn Fn(String)) -> Result<Box<dyn Session>> + Send>;
136
137/// Over Bluetooth, through BlueZ: what the window uses. Tests use a fake.
138fn bluetooth(device: Option<BdAddr>) -> Connect {
139 Box::new(move |found| {
140 let (device, model) = budctl_core::select(budctl_core::scan()?, device)?;
141 found(device.alias.clone());
142 budctl_core::connect(&device, model)
143 })
144}
145
146struct 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
158impl 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)]
329mod 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}