diff --git a/CHANGELOG.md b/CHANGELOG.md index b33da2d6..7c287683 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,24 @@ +# Unreleased + +## Breaking Changes + +- **`Peripheral::notifications()`**: Items are now `Result`, + so lost or unreadable notifications reach the caller as `Err`. +- **`Central::events()`**: Items are now `Result`, so skipped + events reach the caller as `Err(Error::Lagged)`. +- **`Error` enum**: New variant `Lagged(u64)`, the number of items a stream + skipped. Exhaustive matches will need updating. + +## Bugfixes + +- Fix Android `notifications()` spinning at full CPU, unreachable by timeouts, + when the Java stream failed on every poll. +- Fix macOS and Windows `notifications()` silently skipping notifications when + more than 16 were queued. They are reported as `Error::Lagged`. +- Fix macOS, Windows and Android `events()` silently skipping events, + including `DeviceConnected` and `DeviceDisconnected`, when more than 16 were + queued. They are reported as `Error::Lagged`. + # 0.13.4 (2026-10-03) ## Behavior Changes diff --git a/examples/event_driven_discovery.rs b/examples/event_driven_discovery.rs index e16d6fd7..b2fddcc3 100644 --- a/examples/event_driven_discovery.rs +++ b/examples/event_driven_discovery.rs @@ -27,7 +27,7 @@ async fn main() -> anyhow::Result<()> { // Each adapter has an event stream, we fetch via events(), // simplifying the type, this will return what is essentially a - // Future>>. + // Future>>>. let mut events = central.events().await?; // start scanning for devices @@ -36,6 +36,13 @@ async fn main() -> anyhow::Result<()> { // Process events asynchronously. In a real program, run this loop in its own // Tokio task if event handling should proceed independently of other work. while let Some(event) = events.next().await { + let event = match event { + Ok(event) => event, + Err(err) => { + println!("Missed events: {}", err); + continue; + } + }; match event { CentralEvent::DeviceDiscovered(id) => { let peripheral = central.peripheral(&id).await?; diff --git a/examples/subscribe_notify_characteristic.rs b/examples/subscribe_notify_characteristic.rs index 0f763ffc..65a9e312 100644 --- a/examples/subscribe_notify_characteristic.rs +++ b/examples/subscribe_notify_characteristic.rs @@ -78,11 +78,17 @@ async fn main() -> anyhow::Result<()> { let mut notification_stream = peripheral.notifications().await?.take(4); // Process while the BLE connection is not broken or stopped. - while let Some(data) = notification_stream.next().await { - println!( - "Received data from {:?} [{:?}]: {:?}", - local_name, data.uuid, data.value - ); + while let Some(item) = notification_stream.next().await { + match item { + Ok(data) => println!( + "Received data from {:?} [{:?}]: {:?}", + local_name, data.uuid, data.value + ), + Err(err) => println!( + "Lost notifications from {:?}: {}", + local_name, err + ), + } } } } diff --git a/src/api/mod.rs b/src/api/mod.rs index ad7dab7d..9f9ff0c5 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -413,7 +413,12 @@ pub trait Peripheral: Send + Sync + Clone + Debug { /// a notification when a value notification or indication is received from the device. /// The stream will remain valid across connections and can be queried before any connection /// is made. - async fn notifications(&self) -> Result + Send>>>; + /// + /// `Err` items report lost or unreadable notifications, such as + /// [`Error::Lagged`](crate::Error::Lagged). The stream continues. + async fn notifications( + &self, + ) -> Result> + Send>>>; /// Write some data to the descriptor. Returns an error if the write couldn't be sent or (in /// the case of a write-with-response) if the device returns an error. @@ -518,7 +523,10 @@ pub trait Central: Send + Sync + Clone { /// Retrieve a stream of `CentralEvent`s. This stream will receive notifications when events /// occur for this Central module. See [`CentralEvent`] for the full set of possible events. - async fn events(&self) -> Result + Send>>>; + /// + /// `Err` items report skipped events, such as [`Error::Lagged`](crate::Error::Lagged). + /// The stream continues. + async fn events(&self) -> Result> + Send>>>; /// Starts a scan for BLE devices. This scan will generally continue until explicitly stopped, /// although this may depend on your Bluetooth adapter. Discovered devices will be announced diff --git a/src/bluez/adapter.rs b/src/bluez/adapter.rs index 309e28f3..25e0012a 100644 --- a/src/bluez/adapter.rs +++ b/src/bluez/adapter.rs @@ -51,7 +51,7 @@ fn matches_retrieval_options( impl Central for Adapter { type Peripheral = Peripheral; - async fn events(&self) -> Result + Send>>> { + async fn events(&self) -> Result> + Send>>> { // There's a race between getting this event stream and getting the current set of devices. // Get the stream first, on the basis that it's better to have a duplicate DeviceDiscovered // event than to miss one. It's unlikely to happen in any case. @@ -86,7 +86,7 @@ impl Central for Adapter { .filter_map(move |event| central_events(event, session.clone(), adapter_id.clone())) .flat_map(stream::iter); - Ok(Box::pin(initial_events.chain(events))) + Ok(Box::pin(initial_events.chain(events).map(Ok))) } async fn start_scan(&self, filter: ScanFilter) -> Result<()> { diff --git a/src/bluez/peripheral.rs b/src/bluez/peripheral.rs index 5fa134f2..0ed26da9 100644 --- a/src/bluez/peripheral.rs +++ b/src/bluez/peripheral.rs @@ -273,12 +273,14 @@ impl api::Peripheral for Peripheral { } } - async fn notifications(&self) -> Result + Send>>> { + async fn notifications( + &self, + ) -> Result> + Send>>> { let device_id = self.device.clone(); let events = self.session.device_event_stream(&device_id).await?; let services = self.services.clone(); Ok(Box::pin(events.filter_map(move |event| { - ready(value_notification(event, &device_id, services.clone())) + ready(value_notification(event, &device_id, services.clone()).map(Ok)) }))) } diff --git a/src/common/adapter_manager.rs b/src/common/adapter_manager.rs index a22c1095..eb015ba8 100644 --- a/src/common/adapter_manager.rs +++ b/src/common/adapter_manager.rs @@ -11,14 +11,15 @@ // following copyright: // // Copyright (c) 2014 The Rust Project Developers +use super::util::broadcast_stream; +use crate::Result; use crate::api::{CentralEvent, Peripheral}; use crate::platform::PeripheralId; use dashmap::{DashMap, mapref::one::RefMut}; -use futures::stream::{Stream, StreamExt}; +use futures::stream::Stream; use log::trace; use std::pin::Pin; use tokio::sync::broadcast; -use tokio_stream::wrappers::BroadcastStream; #[derive(Debug)] pub struct AdapterManager @@ -53,9 +54,8 @@ where } } - pub fn event_stream(&self) -> Pin + Send>> { - let receiver = self.events_channel.subscribe(); - Box::pin(BroadcastStream::new(receiver).filter_map(|x| async move { x.ok() })) + pub fn event_stream(&self) -> Pin> + Send>> { + broadcast_stream(self.events_channel.subscribe()) } /// Inserts a peripheral if absent and returns the retained instance, @@ -184,7 +184,7 @@ mod tests { async fn notifications( &self, - ) -> Result + Send>>> { + ) -> Result> + Send>>> { unreachable!() } @@ -250,4 +250,27 @@ mod tests { assert_eq!(peripheral.state.load(Ordering::SeqCst), WORKERS); } } + + #[tokio::test] + async fn lagged_event_stream_reports_skipped_count_then_resumes() { + use futures::stream::StreamExt; + + let manager = AdapterManager::::default(); + let events = manager.event_stream(); + let id = TestPeripheral::new().id(); + for _ in 0..20 { + manager.emit(CentralEvent::DeviceUpdated(id.clone())); + } + drop(manager); + + let items: Vec<_> = events.collect().await; + + assert_eq!(items.len(), 17); + assert!(matches!(items[0], Err(crate::Error::Lagged(4)))); + assert!( + items[1..] + .iter() + .all(|item| matches!(item, Ok(CentralEvent::DeviceUpdated(_)))) + ); + } } diff --git a/src/common/util.rs b/src/common/util.rs index 47ef43a8..11480dc7 100644 --- a/src/common/util.rs +++ b/src/common/util.rs @@ -5,15 +5,55 @@ // Licensed under the BSD 3-Clause license. See LICENSE file in the project root // for full license information. -use crate::api::ValueNotification; +use crate::{Error, Result}; use futures::stream::{Stream, StreamExt}; use std::pin::Pin; use tokio::sync::broadcast::Receiver; -use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}; -#[allow(dead_code)] -pub fn notifications_stream_from_broadcast_receiver( - receiver: Receiver, -) -> Pin + Send>> { - Box::pin(BroadcastStream::new(receiver).filter_map(|x| async move { x.ok() })) +/// Streams a broadcast receiver, reporting skipped values as [`Error::Lagged`]. +pub fn broadcast_stream( + receiver: Receiver, +) -> Pin> + Send>> { + Box::pin(BroadcastStream::new(receiver).map(|item| { + item.map_err(|BroadcastStreamRecvError::Lagged(skipped)| Error::Lagged(skipped)) + })) +} + +#[cfg(test)] +mod tests { + use super::broadcast_stream; + use crate::Error; + use crate::api::ValueNotification; + use futures::stream::StreamExt; + use tokio::sync::broadcast; + use uuid::Uuid; + + fn notification(value: u8) -> ValueNotification { + ValueNotification { + uuid: Uuid::nil(), + service_uuid: Uuid::nil(), + value: vec![value], + } + } + + #[tokio::test] + async fn lagged_receiver_reports_skipped_count_then_resumes() { + let (sender, receiver) = broadcast::channel(2); + let stream = broadcast_stream(receiver); + for value in 0..5 { + sender.send(notification(value)).unwrap(); + } + drop(sender); + + let items: Vec<_> = stream.collect().await; + + assert_eq!(items.len(), 3); + assert!(matches!(items[0], Err(Error::Lagged(3)))); + let delivered: Vec = items[1..] + .iter() + .map(|item| item.as_ref().unwrap().value[0]) + .collect(); + assert_eq!(delivered, [3, 4]); + } } diff --git a/src/corebluetooth/adapter.rs b/src/corebluetooth/adapter.rs index b0eb21b2..a141d234 100644 --- a/src/corebluetooth/adapter.rs +++ b/src/corebluetooth/adapter.rs @@ -161,7 +161,7 @@ impl Adapter { impl Central for Adapter { type Peripheral = Peripheral; - async fn events(&self) -> Result + Send>>> { + async fn events(&self) -> Result> + Send>>> { Ok(self.manager.event_stream()) } diff --git a/src/corebluetooth/peripheral.rs b/src/corebluetooth/peripheral.rs index 6c1723c0..b426a319 100644 --- a/src/corebluetooth/peripheral.rs +++ b/src/corebluetooth/peripheral.rs @@ -14,7 +14,7 @@ use crate::{ self, BDAddr, CentralEvent, CharPropFlags, Characteristic, Descriptor, PeripheralProperties, Service, ValueNotification, WriteType, }, - common::{adapter_manager::AdapterManager, util::notifications_stream_from_broadcast_receiver}, + common::{adapter_manager::AdapterManager, util::broadcast_stream}, }; use async_trait::async_trait; use futures::channel::mpsc::{Receiver, SendError, Sender}; @@ -597,9 +597,11 @@ impl api::Peripheral for Peripheral { Ok(()) } - async fn notifications(&self) -> Result + Send>>> { + async fn notifications( + &self, + ) -> Result> + Send>>> { let receiver = self.shared.notifications_channel.subscribe(); - Ok(notifications_stream_from_broadcast_receiver(receiver)) + Ok(broadcast_stream(receiver)) } async fn write_descriptor(&self, descriptor: &Descriptor, data: &[u8]) -> Result<()> { diff --git a/src/droidplug/adapter.rs b/src/droidplug/adapter.rs index af8175fc..9183fc31 100644 --- a/src/droidplug/adapter.rs +++ b/src/droidplug/adapter.rs @@ -133,7 +133,7 @@ impl Central for Adapter { Ok("Android".to_string()) } - async fn events(&self) -> Result + Send>>> { + async fn events(&self) -> Result> + Send>>> { Ok(self.manager.event_stream()) } diff --git a/src/droidplug/peripheral.rs b/src/droidplug/peripheral.rs index 04042913..3f7c3018 100644 --- a/src/droidplug/peripheral.rs +++ b/src/droidplug/peripheral.rs @@ -394,47 +394,43 @@ impl api::Peripheral for Peripheral { .await } - async fn notifications(&self) -> Result + Send>>> { + async fn notifications( + &self, + ) -> Result> + Send>>> { use futures::stream::StreamExt; let shared = self.shared.clone(); let stream = self.with_obj(|env, obj| { let stream = obj.get_notifications(env)?; Ok(JSendStream::new(env, &stream)?) })?; - let stream = stream - .map(move |item| match item { - Ok(item) => { - let vm = jvm()?; - let result: crate::Result<_> = vm - .attach_current_thread(|env| -> jni::errors::Result<_> { - let local_obj = env.new_local_ref(item.as_obj())?; - let characteristic = - env.cast_local::(local_obj)?; - let uuid = characteristic.get_uuid(env)?; - let value = characteristic.get_value(env)?; - let service_uuid = shared - .lock() - .ok() - .and_then(|guard| { - guard - .services - .iter() - .find(|s| s.characteristics.iter().any(|c| c.uuid == uuid)) - .map(|s| s.uuid) - }) - .unwrap_or_default(); - Ok(ValueNotification { - uuid, - service_uuid, - value, - }) + let stream = stream.map(move |item| -> Result { + let item = item?; + jvm()? + .attach_current_thread(|env| -> jni::errors::Result<_> { + let local_obj = env.new_local_ref(item.as_obj())?; + let characteristic = + env.cast_local::(local_obj)?; + let uuid = characteristic.get_uuid(env)?; + let value = characteristic.get_value(env)?; + let service_uuid = shared + .lock() + .ok() + .and_then(|guard| { + guard + .services + .iter() + .find(|s| s.characteristics.iter().any(|c| c.uuid == uuid)) + .map(|s| s.uuid) }) - .map_err(Into::into); - result - } - Err(err) => Err(err.into()), - }) - .filter_map(|item| async { item.ok() }); + .unwrap_or_default(); + Ok(ValueNotification { + uuid, + service_uuid, + value, + }) + }) + .map_err(Into::into) + }); Ok(Box::pin(stream)) } diff --git a/src/lib.rs b/src/lib.rs index 5e05f643..9c417f84 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -150,6 +150,9 @@ pub enum Error { #[error("Runtime Error: {}", _0)] RuntimeError(String), + #[error("Stream lagged, {0} items skipped")] + Lagged(u64), + #[error("{}", _0)] Other(Box), } diff --git a/src/winrtble/adapter.rs b/src/winrtble/adapter.rs index bf9c0882..e6b4377c 100644 --- a/src/winrtble/adapter.rs +++ b/src/winrtble/adapter.rs @@ -197,7 +197,7 @@ impl Debug for Adapter { impl Central for Adapter { type Peripheral = Peripheral; - async fn events(&self) -> Result + Send>>> { + async fn events(&self) -> Result> + Send>>> { Ok(self.manager.event_stream()) } diff --git a/src/winrtble/peripheral.rs b/src/winrtble/peripheral.rs index 4d1b210d..99dc5689 100644 --- a/src/winrtble/peripheral.rs +++ b/src/winrtble/peripheral.rs @@ -23,7 +23,7 @@ use crate::{ Service, ValueNotification, WriteType, bleuuid::{uuid_from_u16, uuid_from_u32}, }, - common::{adapter_manager::AdapterManager, util::notifications_stream_from_broadcast_receiver}, + common::{adapter_manager::AdapterManager, util::broadcast_stream}, }; use async_trait::async_trait; use dashmap::DashMap; @@ -673,9 +673,11 @@ impl ApiPeripheral for Peripheral { ble_characteristic.read_value().await } - async fn notifications(&self) -> Result + Send>>> { + async fn notifications( + &self, + ) -> Result> + Send>>> { let receiver = self.shared.notifications_channel.subscribe(); - Ok(notifications_stream_from_broadcast_receiver(receiver)) + Ok(broadcast_stream(receiver)) } async fn write_descriptor(&self, descriptor: &Descriptor, data: &[u8]) -> Result<()> { diff --git a/tests/common/peripheral_finder.rs b/tests/common/peripheral_finder.rs index 3ece22d0..246fa26c 100644 --- a/tests/common/peripheral_finder.rs +++ b/tests/common/peripheral_finder.rs @@ -201,17 +201,31 @@ pub fn find_characteristic( .unwrap_or_else(|| panic!("characteristic {} not found", uuid)) } +/// The peripheral's notification stream, failing the test on an `Err` item. +pub async fn notification_stream( + peripheral: &Peripheral, +) -> impl futures::Stream + Unpin { + use futures::StreamExt; + + peripheral + .notifications() + .await + .unwrap() + .map(|item| item.expect("notification stream reported an error")) +} + /// Subscribe to `adapter.events()` and forward every event into an unbounded /// channel. /// /// The central event broadcast channel (`src/common/adapter_manager.rs`) has -/// capacity 16 and `event_stream()` silently drops events for a receiver that -/// falls behind (`filter_map(|x| x.ok())`). During an active scan CoreBluetooth -/// emits `DeviceUpdated` for every nearby advertiser, so a receiver that isn't -/// polled continuously can lose events (in particular `DeviceConnected`) -/// between the moment it subscribes and the moment a test gets around to -/// reading it. Forwarding into an unbounded channel from a task that starts -/// draining immediately avoids that loss. The subscription (`events().await`) +/// capacity 16, and a receiver that falls behind loses events and gets +/// `Error::Lagged`. During an active scan CoreBluetooth emits `DeviceUpdated` +/// for every nearby advertiser, so a receiver that isn't polled continuously +/// can lose events (in particular `DeviceConnected`) between the moment it +/// subscribes and the moment a test gets around to reading it. Forwarding into +/// an unbounded channel from a task that starts draining immediately avoids +/// that loss, and a lag that still happens panics the collector, which closes +/// the channel for `wait_for_event`. The subscription (`events().await`) /// happens before the forwarding task is spawned, so no event between the /// caller's request and the task starting is missed. pub async fn spawn_event_collector() -> UnboundedReceiver { @@ -228,11 +242,12 @@ pub async fn spawn_event_collector() -> UnboundedReceiver { tokio::select! { _ = tx.closed() => break, ev = events.next() => match ev { - Some(e) => { + Some(Ok(e)) => { if tx.send(e).is_err() { break; } } + Some(Err(err)) => panic!("adapter event stream reported an error: {err}"), None => break, }, } diff --git a/tests/common/test_cases.rs b/tests/common/test_cases.rs index 29c975da..105401d2 100644 --- a/tests/common/test_cases.rs +++ b/tests/common/test_cases.rs @@ -114,7 +114,7 @@ pub async fn test_clear_peripherals_rediscovers_device() { adapter.start_scan(ScanFilter::default()).await.unwrap(); time::timeout(Duration::from_secs(15), async { loop { - if matches!(events.next().await, Some(CentralEvent::DeviceDiscovered(id)) if id == peripheral_id) + if matches!(events.next().await, Some(Ok(CentralEvent::DeviceDiscovered(id))) if id == peripheral_id) { break; } @@ -285,6 +285,8 @@ pub async fn test_advertisement_manufacturer_data() { loop { tokio::select! { Some(event) = events.next() => { + // Advertisements repeat, so a skipped one is seen again. + let Ok(event) = event else { continue }; if let CentralEvent::ManufacturerDataAdvertisement { manufacturer_data, .. } = event { if manufacturer_data.contains_key(&gatt_uuids::MANUFACTURER_COMPANY_ID) { found_manufacturer_data = true; @@ -320,6 +322,8 @@ pub async fn test_advertisement_services() { loop { tokio::select! { Some(event) = events.next() => { + // Advertisements repeat, so a skipped one is seen again. + let Ok(event) = event else { continue }; if let CentralEvent::ServicesAdvertisement { services, .. } = event { if services.contains(&gatt_uuids::CONTROL_SERVICE) { found_services = true; @@ -934,7 +938,7 @@ pub async fn test_subscribe_and_receive_notifications() { let peripheral = peripheral_finder::find_and_connect().await; peripheral_finder::reset_peripheral(&peripheral).await; let char = peripheral_finder::find_characteristic(&peripheral, gatt_uuids::NOTIFY_CHAR); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral.subscribe(&char).await.unwrap(); peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; @@ -978,7 +982,7 @@ pub async fn test_subscribe_and_receive_indications() { let peripheral = peripheral_finder::find_and_connect().await; peripheral_finder::reset_peripheral(&peripheral).await; let char = peripheral_finder::find_characteristic(&peripheral, gatt_uuids::INDICATE_CHAR); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral.subscribe(&char).await.unwrap(); peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; @@ -1019,7 +1023,7 @@ pub async fn test_unsubscribe_stops_notifications() { let peripheral = peripheral_finder::find_and_connect().await; peripheral_finder::reset_peripheral(&peripheral).await; let char = peripheral_finder::find_characteristic(&peripheral, gatt_uuids::NOTIFY_CHAR); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral.subscribe(&char).await.unwrap(); peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; @@ -1098,7 +1102,7 @@ pub async fn test_configurable_notification_payload() { .await .unwrap(); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral.subscribe(&config_char).await.unwrap(); peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; @@ -1160,7 +1164,7 @@ pub async fn test_mtu_sized_notification_payload() { .await .unwrap(); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral.subscribe(&config_char).await.unwrap(); peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; @@ -1217,7 +1221,7 @@ pub async fn test_resubscribe_does_not_duplicate_notifications() { // Take the notification stream before starting notifications so nothing // is missed. - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; peripheral_finder::send_control_command(&peripheral, gatt_uuids::CMD_START_NOTIFICATIONS).await; let mut received = Vec::new(); @@ -1288,7 +1292,7 @@ pub async fn test_refused_subscribe_returns_error() { ); let notify_char = peripheral_finder::find_characteristic(&peripheral, gatt_uuids::NOTIFY_CHAR); - let mut stream = peripheral.notifications().await.unwrap(); + let mut stream = peripheral_finder::notification_stream(&peripheral).await; tokio::time::timeout(Duration::from_secs(10), peripheral.subscribe(¬ify_char)) .await .expect("subscribe(NOTIFY_CHAR) timed out")