Krolyxon
Commit 47dc78a

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

364 lines · 13.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 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}
48
49#[derive(Debug, Clone)]
50pub enum Event {
51 /// The worker takes commands here.
52 Ready(Sender<Command>),
53 /// The buds a new session is for, by the name their owner gave them. Answers no command.
54 Found(String),
55 /// The answer to [`Command::Read`]. Boxed: it is much larger than the other events.
56 Read(Box<Reading>),
57 /// The answer to [`Command::Noise`]: what the buds report after the write.
58 Noise(NoiseControl),
59 /// The answer to [`Command::Switch`]: the switch, as read back.
60 Switch(Switch),
61 /// The answer to [`Command::Eq`]: the preset now in use.
62 Eq(String),
63 /// The answer to [`Command::CustomEq`]: the preset, as read back.
64 CustomEq(EqPreset),
65 /// The answer to [`Command::Touch`]: the whole gesture table, as read back.
66 Touch(Vec<Gesture>),
67 /// The answer to [`Command::HoldCycle`]: the modes, as read back.
68 HoldCycle(Vec<String>),
69 /// The answer to a command that failed. The session is closed; the next command opens a
70 /// new one.
71 Failed { kind: Kind, message: String },
72 /// The link dropped while the session waited. Answers no command.
73 Lost(String),
74}
75
76#[derive(Debug, Clone)]
77pub struct Reading {
78 pub status: Status,
79 /// The EQ presets, when the buds can list them: they name the custom ones.
80 pub eq: Option<EqPresets>,
81 /// The touch controls, when the buds report them.
82 pub touch: Option<TouchControls>,
83}
84
85/// The worker, as a stream of events for an iced subscription. `device` picks the buds, as
86/// `budctl --device` does.
87pub fn run(device: &Option<BdAddr>) -> impl Stream<Item = Event> + use<> {
88 let device = *device;
89 iced::stream::channel(16, async move |mut output: channel::Sender<Event>| {
90 let (commands, inbox) = mpsc::channel();
91 let (events, mut outbox) = channel::unbounded();
92 if output.send(Event::Ready(commands)).await.is_err() {
93 return;
94 }
95 let worker = Worker::new(bluetooth(device), events, IDLE);
96 let started = thread::Builder::new()
97 .name("budctl-buds".to_owned())
98 .spawn(move || worker.run(&inbox));
99 if let Err(e) = started {
100 let message = format!("could not start the Bluetooth thread: {e}");
101 let _ = output.send(Event::Lost(message)).await;
102 return;
103 }
104 while let Some(event) = outbox.next().await {
105 if output.send(event).await.is_err() {
106 break;
107 }
108 }
109 })
110}
111
112/// Finds the buds and opens a session, calling `found` with their name once it is known.
113type Connect = Box<dyn FnMut(&dyn Fn(String)) -> Result<Box<dyn Session>> + Send>;
114
115/// Over Bluetooth, through BlueZ: what the window uses. Tests use a fake.
116fn bluetooth(device: Option<BdAddr>) -> Connect {
117 Box::new(move |found| {
118 let (device, model) = budctl_core::select(budctl_core::scan()?, device)?;
119 found(device.alias.clone());
120 budctl_core::connect(&device, model)
121 })
122}
123
124struct Worker {
125 connect: Connect,
126 session: Option<Box<dyn Session>>,
127 events: channel::UnboundedSender<Event>,
128 last_command: Instant,
129 /// How long the session stays open after the last command.
130 idle: Duration,
131}
132
133impl Worker {
134 fn new(connect: Connect, events: channel::UnboundedSender<Event>, idle: Duration) -> Self {
135 Worker {
136 connect,
137 session: None,
138 events,
139 last_command: Instant::now(),
140 idle,
141 }
142 }
143
144 /// Runs until the window is gone.
145 fn run(mut self, inbox: &Receiver<Command>) {
146 loop {
147 let next = if self.session.is_some() {
148 inbox.recv_timeout(SERVE_EVERY)
149 } else {
150 inbox.recv().map_err(|_| RecvTimeoutError::Disconnected)
151 };
152 match next {
153 Ok(command) => {
154 self.last_command = Instant::now();
155 self.handle(command);
156 }
157 Err(RecvTimeoutError::Timeout) => self.wait(),
158 Err(RecvTimeoutError::Disconnected) => return,
159 }
160 }
161 }
162
163 /// Between commands: answer the buds, and close the session once it has been idle long
164 /// enough.
165 fn wait(&mut self) {
166 let Some(session) = self.session.as_mut() else {
167 return;
168 };
169 if self.last_command.elapsed() >= self.idle {
170 self.session = None;
171 if budctl_core::debug::enabled() {
172 eprintln!(
173 "# session closed after {} s without a command",
174 self.idle.as_secs()
175 );
176 }
177 } else if let Err(error) = session.serve() {
178 self.session = None;
179 self.send(Event::Lost(error.to_string()));
180 }
181 }
182
183 fn handle(&mut self, command: Command) {
184 let answer = match command {
185 Command::Read => self.read().map(|reading| Event::Read(Box::new(reading))),
186 Command::Noise(mode) => self
187 .session()
188 .and_then(|session| session.set_noise_control(mode))
189 .map(Event::Noise),
190 Command::Switch(key, on) => self
191 .session()
192 .and_then(|session| session.set_switch(key, on))
193 .map(Event::Switch),
194 Command::Eq(preset) => self
195 .session()
196 .and_then(|session| session.set_eq(&preset))
197 .map(Event::Eq),
198 Command::CustomEq(preset, bands) => self
199 .session()
200 .and_then(|session| session.set_custom_eq(&preset, &bands))
201 .map(Event::CustomEq),
202 Command::Touch { side, gesture, action } => self
203 .session()
204 .and_then(|session| session.set_touch(&side, &gesture, &action, false))
205 .map(|change| Event::Touch(change.gestures)),
206 Command::HoldCycle(modes) => {
207 let modes: Vec<&str> = modes.iter().map(String::as_str).collect();
208 self.session()
209 .and_then(|session| session.set_hold_cycle(&modes))
210 .map(Event::HoldCycle)
211 }
212 };
213 match answer {
214 Ok(event) => self.send(event),
215 Err(error) => {
216 self.session = None;
217 self.send(Event::Failed {
218 kind: error.kind(),
219 message: error.to_string(),
220 });
221 }
222 }
223 }
224
225 fn read(&mut self) -> Result<Reading> {
226 let session = self.session()?;
227 let status = session.status()?;
228 let eq = session.eq_presets().ok();
229 let touch = session.touch_controls().ok();
230 Ok(Reading { status, eq, touch })
231 }
232
233 /// The open session, opening one first if there is none.
234 fn session(&mut self) -> Result<&mut Box<dyn Session>> {
235 if self.session.is_none() {
236 let events = self.events.clone();
237 let found = move |name| {
238 let _ = events.unbounded_send(Event::Found(name));
239 };
240 self.session = Some((self.connect)(&found)?);
241 }
242 self.session
243 .as_mut()
244 .ok_or_else(|| Error::NotConnected("the session closed".to_owned()))
245 }
246
247 fn send(&self, event: Event) {
248 // An error only means the window is gone; `run` ends on its next receive.
249 let _ = self.events.unbounded_send(event);
250 }
251}
252
253#[cfg(test)]
254mod tests {
255 use super::*;
256 use iced::futures::executor::block_on;
257 use std::sync::Arc;
258 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
259
260 /// What the fake sessions report back to the test.
261 #[derive(Clone, Default)]
262 struct Probe {
263 opened: Arc<AtomicUsize>,
264 served: Arc<AtomicUsize>,
265 closed: Arc<AtomicBool>,
266 /// Make `serve` fail, as when the buds drop the link.
267 drop_link: Arc<AtomicBool>,
268 }
269
270 struct Fake(Probe);
271
272 impl Session for Fake {
273 fn status(&mut self) -> Result<Status> {
274 Ok(Status::default())
275 }
276
277 fn serve(&mut self) -> Result<()> {
278 self.0.served.fetch_add(1, Ordering::SeqCst);
279 if self.0.drop_link.load(Ordering::SeqCst) {
280 return Err(Error::NotConnected("the buds closed the control link".to_owned()));
281 }
282 Ok(())
283 }
284 }
285
286 impl Drop for Fake {
287 fn drop(&mut self) {
288 self.0.closed.store(true, Ordering::SeqCst);
289 }
290 }
291
292 /// A worker on its own thread with fake buds, closing idle sessions after 200 ms.
293 fn start(
294 probe: &Probe,
295 ) -> (
296 Sender<Command>,
297 channel::UnboundedReceiver<Event>,
298 thread::JoinHandle<()>,
299 ) {
300 let fake = probe.clone();
301 let connect: Connect = Box::new(move |found| {
302 found("Fake buds".to_owned());
303 fake.opened.fetch_add(1, Ordering::SeqCst);
304 fake.closed.store(false, Ordering::SeqCst);
305 Ok(Box::new(Fake(fake.clone())))
306 });
307 let (events, outbox) = channel::unbounded();
308 let (commands, inbox) = mpsc::channel();
309 let worker = Worker::new(connect, events, Duration::from_millis(200));
310 let thread = thread::spawn(move || worker.run(&inbox));
311 (commands, outbox, thread)
312 }
313
314 /// Waits, at most 5 s, for `done`. A receive that times out stands in for a sleep.
315 fn wait_for(done: impl Fn() -> bool) {
316 let (_keep, never) = mpsc::channel::<()>();
317 let deadline = Instant::now() + Duration::from_secs(5);
318 while !done() {
319 assert!(Instant::now() < deadline, "gave up waiting");
320 let _ = never.recv_timeout(Duration::from_millis(10));
321 }
322 }
323
324 #[test]
325 fn an_idle_session_closes_and_the_next_command_opens_another() {
326 let probe = Probe::default();
327 let (commands, mut events, thread) = start(&probe);
328
329 commands.send(Command::Read).unwrap();
330 assert!(matches!(block_on(events.next()), Some(Event::Found(name)) if name == "Fake buds"));
331 assert!(matches!(block_on(events.next()), Some(Event::Read(_))));
332 assert_eq!(probe.opened.load(Ordering::SeqCst), 1);
333
334 // While it waits, it answers the buds; then it closes the session.
335 wait_for(|| probe.closed.load(Ordering::SeqCst));
336 assert!(probe.served.load(Ordering::SeqCst) > 0);
337
338 commands.send(Command::Read).unwrap();
339 assert!(matches!(block_on(events.next()), Some(Event::Found(_))));
340 assert!(matches!(block_on(events.next()), Some(Event::Read(_))));
341 assert_eq!(probe.opened.load(Ordering::SeqCst), 2);
342
343 // The window goes away: the worker ends.
344 drop(commands);
345 thread.join().unwrap();
346 }
347
348 #[test]
349 fn a_dropped_link_is_reported_once() {
350 let probe = Probe::default();
351 let (commands, mut events, thread) = start(&probe);
352 commands.send(Command::Read).unwrap();
353 let _ = block_on(events.next());
354 let _ = block_on(events.next());
355
356 probe.drop_link.store(true, Ordering::SeqCst);
357 assert!(matches!(block_on(events.next()), Some(Event::Lost(_))));
358 assert!(probe.closed.load(Ordering::SeqCst));
359
360 drop(commands);
361 thread.join().unwrap();
362 assert!(block_on(events.next()).is_none(), "nothing more after the link dropped");
363 }
364}