Krolyxon
Commit e8c78b9

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

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