feat(api): Yield Result items from Central::events()

CHANGELOG.md

@@ -1,3 +1,24 @@
+# Unreleased
+
+## Breaking Changes
+
+- **`Peripheral::notifications()`**: Items are now `Result<ValueNotification>`,
+  so lost or unreadable notifications reach the caller as `Err`.
+- **`Central::events()`**: Items are now `Result<CentralEvent>`, 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

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<Result<Stream<Item=CentralEvent>>>.
+    // Future<Result<Stream<Item=Result<CentralEvent>>>>.
     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?;

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
+                                        ),
+                                    }
                                 }
                             }
                         }

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<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>>;
+    ///
+    /// `Err` items report lost or unreadable notifications, such as
+    /// [`Error::Lagged`](crate::Error::Lagged). The stream continues.
+    async fn notifications(
+        &self,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ValueNotification>> + 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<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>>;
+    ///
+    /// `Err` items report skipped events, such as [`Error::Lagged`](crate::Error::Lagged).
+    /// The stream continues.
+    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = Result<CentralEvent>> + 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

src/bluez/adapter.rs

@@ -51,7 +51,7 @@ fn matches_retrieval_options(
 impl Central for Adapter {
     type Peripheral = Peripheral;
 
-    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>> {
+    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = Result<CentralEvent>> + 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<()> {

src/bluez/peripheral.rs

@@ -273,12 +273,14 @@ impl api::Peripheral for Peripheral {
         }
     }
 
-    async fn notifications(&self) -> Result<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>> {
+    async fn notifications(
+        &self,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ValueNotification>> + 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))
         })))
     }
 

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<PeripheralType>
@@ -53,9 +54,8 @@ where
         }
     }
 
-    pub fn event_stream(&self) -> Pin<Box<dyn Stream<Item = CentralEvent> + 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<Box<dyn Stream<Item = Result<CentralEvent>> + 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<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>> {
+        ) -> Result<Pin<Box<dyn Stream<Item = Result<ValueNotification>> + 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::<TestPeripheral>::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(_))))
+        );
+    }
 }

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<ValueNotification>,
-) -> Pin<Box<dyn Stream<Item = ValueNotification> + 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<T: Clone + Send + 'static>(
+    receiver: Receiver<T>,
+) -> Pin<Box<dyn Stream<Item = Result<T>> + 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<u8> = items[1..]
+            .iter()
+            .map(|item| item.as_ref().unwrap().value[0])
+            .collect();
+        assert_eq!(delivered, [3, 4]);
+    }
 }

src/corebluetooth/adapter.rs

@@ -161,7 +161,7 @@ impl Adapter {
 impl Central for Adapter {
     type Peripheral = Peripheral;
 
-    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>> {
+    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = Result<CentralEvent>> + Send>>> {
         Ok(self.manager.event_stream())
     }
 

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<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>> {
+    async fn notifications(
+        &self,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ValueNotification>> + 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<()> {

src/droidplug/adapter.rs

@@ -133,7 +133,7 @@ impl Central for Adapter {
         Ok("Android".to_string())
     }
 
-    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>> {
+    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = Result<CentralEvent>> + Send>>> {
         Ok(self.manager.event_stream())
     }
 

src/droidplug/peripheral.rs

@@ -394,47 +394,43 @@ impl api::Peripheral for Peripheral {
             .await
     }
 
-    async fn notifications(&self) -> Result<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>> {
+    async fn notifications(
+        &self,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ValueNotification>> + 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::<JBluetoothGattCharacteristic>(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<ValueNotification> {
+            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::<JBluetoothGattCharacteristic>(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))
     }
 

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<dyn std::error::Error + Send + Sync>),
 }

src/winrtble/adapter.rs

@@ -197,7 +197,7 @@ impl Debug for Adapter {
 impl Central for Adapter {
     type Peripheral = Peripheral;
 
-    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>> {
+    async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = Result<CentralEvent>> + Send>>> {
         Ok(self.manager.event_stream())
     }