Krolyxon
Commit 1978b2f

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

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