Krolyxon
Commit a21a864

Branches

Refs →

budctlcratesbudctl-guisrcworker.rs

183 lines · 6.5 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::{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}
33
34#[derive(Debug, Clone)]
35pub enum Event {
36 /// The worker takes commands here.
37 Ready(Sender<Command>),
38 /// The buds a new session is for, by the name their owner gave them. Answers no command.
39 Found(String),
40 /// The answer to [`Command::Read`]. Boxed: it is much larger than the other events.
41 Read(Box<Reading>),
42 /// The answer to [`Command::Noise`]: what the buds report after the write.
43 Noise(NoiseControl),
44 /// The answer to [`Command::Switch`]: the switch, as read back.
45 Switch(Switch),
46 /// The answer to a command that failed. The session is closed; the next command opens a
47 /// new one.
48 Failed { kind: Kind, message: String },
49 /// The link dropped while the session waited. Answers no command.
50 Lost(String),
51}
52
53#[derive(Debug, Clone)]
54pub struct Reading {
55 pub status: Status,
56 /// The EQ presets, when the buds can list them: they name the custom ones.
57 pub eq: Option<EqPresets>,
58}
59
60/// The worker, as a stream of events for an iced subscription. `device` picks the buds, as
61/// `budctl --device` does.
62pub fn run(device: &Option<BdAddr>) -> impl Stream<Item = Event> + use<> {
63 let device = *device;
64 iced::stream::channel(16, async move |mut output: channel::Sender<Event>| {
65 let (commands, inbox) = mpsc::channel();
66 let (events, mut outbox) = channel::unbounded();
67 if output.send(Event::Ready(commands)).await.is_err() {
68 return;
69 }
70 let worker = Worker {
71 device,
72 session: None,
73 events,
74 last_command: Instant::now(),
75 };
76 let started = thread::Builder::new()
77 .name("budctl-buds".to_owned())
78 .spawn(move || worker.run(&inbox));
79 if let Err(e) = started {
80 let message = format!("could not start the Bluetooth thread: {e}");
81 let _ = output.send(Event::Lost(message)).await;
82 return;
83 }
84 while let Some(event) = outbox.next().await {
85 if output.send(event).await.is_err() {
86 break;
87 }
88 }
89 })
90}
91
92struct Worker {
93 device: Option<BdAddr>,
94 session: Option<Box<dyn Session>>,
95 events: channel::UnboundedSender<Event>,
96 last_command: Instant,
97}
98
99impl Worker {
100 /// Runs until the window is gone.
101 fn run(mut self, inbox: &Receiver<Command>) {
102 loop {
103 let next = if self.session.is_some() {
104 inbox.recv_timeout(SERVE_EVERY)
105 } else {
106 inbox.recv().map_err(|_| RecvTimeoutError::Disconnected)
107 };
108 match next {
109 Ok(command) => {
110 self.last_command = Instant::now();
111 self.handle(command);
112 }
113 Err(RecvTimeoutError::Timeout) => self.wait(),
114 Err(RecvTimeoutError::Disconnected) => return,
115 }
116 }
117 }
118
119 /// Between commands: answer the buds, and close the session once it has been idle long
120 /// enough.
121 fn wait(&mut self) {
122 let Some(session) = self.session.as_mut() else {
123 return;
124 };
125 if self.last_command.elapsed() >= IDLE {
126 self.session = None;
127 if budctl_core::debug::enabled() {
128 eprintln!("# session closed after {} s without a command", IDLE.as_secs());
129 }
130 } else if let Err(error) = session.serve() {
131 self.session = None;
132 self.send(Event::Lost(error.to_string()));
133 }
134 }
135
136 fn handle(&mut self, command: Command) {
137 let answer = match command {
138 Command::Read => self.read().map(|reading| Event::Read(Box::new(reading))),
139 Command::Noise(mode) => self
140 .session()
141 .and_then(|session| session.set_noise_control(mode))
142 .map(Event::Noise),
143 Command::Switch(key, on) => self
144 .session()
145 .and_then(|session| session.set_switch(key, on))
146 .map(Event::Switch),
147 };
148 match answer {
149 Ok(event) => self.send(event),
150 Err(error) => {
151 self.session = None;
152 self.send(Event::Failed {
153 kind: error.kind(),
154 message: error.to_string(),
155 });
156 }
157 }
158 }
159
160 fn read(&mut self) -> Result<Reading> {
161 let session = self.session()?;
162 let status = session.status()?;
163 let eq = session.eq_presets().ok();
164 Ok(Reading { status, eq })
165 }
166
167 /// The open session, opening one first if there is none.
168 fn session(&mut self) -> Result<&mut Box<dyn Session>> {
169 if self.session.is_none() {
170 let (device, model) = budctl_core::select(budctl_core::scan()?, self.device)?;
171 self.send(Event::Found(device.alias.clone()));
172 self.session = Some(budctl_core::connect(&device, model)?);
173 }
174 self.session
175 .as_mut()
176 .ok_or_else(|| Error::NotConnected("the session closed".to_owned()))
177 }
178
179 fn send(&self, event: Event) {
180 // An error only means the window is gone; `run` ends on its next receive.
181 let _ = self.events.unbounded_send(event);
182 }
183}