diff --git a/src/lib.rs b/src/lib.rs index 3469815..12f86b6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,10 +24,18 @@ fn callbacks_guard(callbacks: &Mutex) -> MutexGuard<'_, Callbacks> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn watch_task_guard( + watch_task: &Mutex>>, +) -> MutexGuard<'_, Option>> { + watch_task + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + #[napi] pub struct Emitter { callbacks: Arc>, - watch_task: Option>, + watch_task: Arc>>>, } #[napi] @@ -44,16 +52,21 @@ impl Emitter { })); Self { callbacks, - watch_task: None, + watch_task: Arc::new(Mutex::new(None)), } } - async fn start_watching(&mut self) -> Result<()> { - if matches!(self.watch_task.as_ref(), Some(task) if !task.is_finished()) { + // addAttach and addDetach run at the same time, so one lock has to cover + // both the check and the store. Release it in between and each starts a + // watcher, which then reports every plug twice. + fn start_watching(&self) -> Result<()> { + let mut watch_task = watch_task_guard(&self.watch_task); + + if matches!(watch_task.as_ref(), Some(task) if !task.is_finished()) { return Ok(()); } - self.watch_task = None; + *watch_task = None; let callbacks = self.callbacks.clone(); let mut watch_stream = match nusb::watch_devices() { Ok(watch_stream) => watch_stream, @@ -64,7 +77,7 @@ impl Emitter { } }; - self.watch_task = Some(tokio::spawn(async move { + *watch_task = Some(tokio::spawn(async move { while let Some(ev) = watch_stream.next().await { match ev { HotplugEvent::Connected(info) => { @@ -89,61 +102,62 @@ impl Emitter { Ok(()) } - async fn stop_watching(&mut self) { - let has_listeners = { - let cb = self.callbacks(); - cb.attach.is_some() || cb.detach.is_some() - }; + fn stop_watching(&self) { + // Keep the callbacks lock for the whole decision. Drop it after the + // check and a listener registered in the gap would find its watcher + // aborted here, leaving nothing watching. + let callbacks = self.callbacks(); + let has_listeners = callbacks.attach.is_some() || callbacks.detach.is_some(); if !has_listeners { - if let Some(task) = self.watch_task.take() { + if let Some(task) = watch_task_guard(&self.watch_task).take() { task.abort(); } } } #[napi] - pub async unsafe fn addAttach( - &mut self, + pub async fn addAttach( + &self, callback: ThreadsafeFunction, ) -> Result<()> { { self.callbacks().attach = Some(callback); } - self.start_watching().await + self.start_watching() } #[napi] - pub async unsafe fn removeAttach(&mut self) { + pub async fn removeAttach(&self) { { self.callbacks().attach = None; } - self.stop_watching().await; + self.stop_watching(); } #[napi] - pub async unsafe fn addDetach( - &mut self, + pub async fn addDetach( + &self, callback: ThreadsafeFunction, ) -> Result<()> { { self.callbacks().detach = Some(callback); } - self.start_watching().await + self.start_watching() } #[napi] - pub async unsafe fn removeDetach(&mut self) { + pub async fn removeDetach(&self) { { self.callbacks().detach = None; } - self.stop_watching().await; + self.stop_watching(); } } impl Drop for Emitter { fn drop(&mut self) { - if let Some(task) = self.watch_task.take() { + if let Some(task) = watch_task_guard(&self.watch_task).take() { task.abort(); } }