| @@ -76,12 +76,7 @@ pub fn run(device: &Option<BdAddr>) -> impl Stream<Item = Event> + use<> { |
| 76 | 76 | | if output.send(Event::Ready(commands)).await.is_err() { |
| 77 | 77 | | return; |
| 78 | 78 | | } |
| 79 | | − | let worker = Worker { |
| 80 | | − | device, |
| 81 | | − | session: None, |
| 82 | | − | events, |
| 83 | | − | last_command: Instant::now(), |
| 84 | | − | }; |
| 79 | + | let worker = Worker::new(bluetooth(device), events, IDLE); |
| 85 | 80 | | let started = thread::Builder::new() |
| 86 | 81 | | .name("budctl-buds".to_owned()) |
| 87 | 82 | | .spawn(move || worker.run(&inbox)); |
| @@ -98,14 +93,38 @@ pub fn run(device: &Option<BdAddr>) -> impl Stream<Item = Event> + use<> { |
| 98 | 93 | | }) |
| 99 | 94 | | } |
| 100 | 95 | | |
| 96 | + | /// Finds the buds and opens a session, calling `found` with their name once it is known. |
| 97 | + | type 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. |
| 100 | + | fn 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 | + | |
| 101 | 108 | | struct Worker { |
| 102 | | − | device: Option<BdAddr>, |
| 109 | + | connect: Connect, |
| 103 | 110 | | session: Option<Box<dyn Session>>, |
| 104 | 111 | | events: channel::UnboundedSender<Event>, |
| 105 | 112 | | last_command: Instant, |
| 113 | + | /// How long the session stays open after the last command. |
| 114 | + | idle: Duration, |
| 106 | 115 | | } |
| 107 | 116 | | |
| 108 | 117 | | impl 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 | + | |
| 109 | 128 | | /// Runs until the window is gone. |
| 110 | 129 | | fn run(mut self, inbox: &Receiver<Command>) { |
| 111 | 130 | | loop { |
| @@ -131,10 +150,13 @@ impl Worker { |
| 131 | 150 | | let Some(session) = self.session.as_mut() else { |
| 132 | 151 | | return; |
| 133 | 152 | | }; |
| 134 | | − | if self.last_command.elapsed() >= IDLE { |
| 153 | + | if self.last_command.elapsed() >= self.idle { |
| 135 | 154 | | self.session = None; |
| 136 | 155 | | if budctl_core::debug::enabled() { |
| 137 | | − | eprintln!("# session closed after {} s without a command", IDLE.as_secs()); |
| 156 | + | eprintln!( |
| 157 | + | "# session closed after {} s without a command", |
| 158 | + | self.idle.as_secs() |
| 159 | + | ); |
| 138 | 160 | | } |
| 139 | 161 | | } else if let Err(error) = session.serve() { |
| 140 | 162 | | self.session = None; |
| @@ -184,9 +206,11 @@ impl Worker { |
| 184 | 206 | | /// The open session, opening one first if there is none. |
| 185 | 207 | | fn session(&mut self) -> Result<&mut Box<dyn Session>> { |
| 186 | 208 | | 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)?); |
| 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)?); |
| 190 | 214 | | } |
| 191 | 215 | | self.session |
| 192 | 216 | | .as_mut() |
| @@ -198,3 +222,116 @@ impl Worker { |
| 198 | 222 | | let _ = self.events.unbounded_send(event); |
| 199 | 223 | | } |
| 200 | 224 | | } |
| 225 | + | |
| 226 | + | #[cfg(test)] |
| 227 | + | mod 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 | + | } |