Krolyxon
Commit 904ab9a

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

200 lines · 7.2 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 {
80 device,
81 session: None,
82 events,
83 last_command: Instant::now(),
84 };
85 let started = thread::Builder::new()
86 .name("budctl-buds".to_owned())
87 .spawn(move || worker.run(&inbox));
88 if let Err(e) = started {
89 let message = format!("could not start the Bluetooth thread: {e}");
90 let _ = output.send(Event::Lost(message)).await;
91 return;
92 }
93 while let Some(event) = outbox.next().await {
94 if output.send(event).await.is_err() {
95 break;
96 }
97 }
98 })
99}
100
101struct Worker {
102 device: Option<BdAddr>,
103 session: Option<Box<dyn Session>>,
104 events: channel::UnboundedSender<Event>,
105 last_command: Instant,
106}
107
108impl Worker {
109 /// Runs until the window is gone.
110 fn run(mut self, inbox: &Receiver<Command>) {
111 loop {
112 let next = if self.session.is_some() {
113 inbox.recv_timeout(SERVE_EVERY)
114 } else {
115 inbox.recv().map_err(|_| RecvTimeoutError::Disconnected)
116 };
117 match next {
118 Ok(command) => {
119 self.last_command = Instant::now();
120 self.handle(command);
121 }
122 Err(RecvTimeoutError::Timeout) => self.wait(),
123 Err(RecvTimeoutError::Disconnected) => return,
124 }
125 }
126 }
127
128 /// Between commands: answer the buds, and close the session once it has been idle long
129 /// enough.
130 fn wait(&mut self) {
131 let Some(session) = self.session.as_mut() else {
132 return;
133 };
134 if self.last_command.elapsed() >= IDLE {
135 self.session = None;
136 if budctl_core::debug::enabled() {
137 eprintln!("# session closed after {} s without a command", IDLE.as_secs());
138 }
139 } else if let Err(error) = session.serve() {
140 self.session = None;
141 self.send(Event::Lost(error.to_string()));
142 }
143 }
144
145 fn handle(&mut self, command: Command) {
146 let answer = match command {
147 Command::Read => self.read().map(|reading| Event::Read(Box::new(reading))),
148 Command::Noise(mode) => self
149 .session()
150 .and_then(|session| session.set_noise_control(mode))
151 .map(Event::Noise),
152 Command::Switch(key, on) => self
153 .session()
154 .and_then(|session| session.set_switch(key, on))
155 .map(Event::Switch),
156 Command::Eq(preset) => self
157 .session()
158 .and_then(|session| session.set_eq(&preset))
159 .map(Event::Eq),
160 Command::CustomEq(preset, bands) => self
161 .session()
162 .and_then(|session| session.set_custom_eq(&preset, &bands))
163 .map(Event::CustomEq),
164 };
165 match answer {
166 Ok(event) => self.send(event),
167 Err(error) => {
168 self.session = None;
169 self.send(Event::Failed {
170 kind: error.kind(),
171 message: error.to_string(),
172 });
173 }
174 }
175 }
176
177 fn read(&mut self) -> Result<Reading> {
178 let session = self.session()?;
179 let status = session.status()?;
180 let eq = session.eq_presets().ok();
181 Ok(Reading { status, eq })
182 }
183
184 /// The open session, opening one first if there is none.
185 fn session(&mut self) -> Result<&mut Box<dyn Session>> {
186 if self.session.is_none() {
187 let (device, model) = budctl_core::select(budctl_core::scan()?, self.device)?;
188 self.send(Event::Found(device.alias.clone()));
189 self.session = Some(budctl_core::connect(&device, model)?);
190 }
191 self.session
192 .as_mut()
193 .ok_or_else(|| Error::NotConnected("the session closed".to_owned()))
194 }
195
196 fn send(&self, event: Event) {
197 // An error only means the window is gone; `run` ends on its next receive.
198 let _ = self.events.unbounded_send(event);
199 }
200}