[fp-rs] Updating device scanning interface, split into start/stop/next.

fastpair/rust/src/bluetooth/common.rs

@@ -12,10 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::pin::Pin;
-
 use async_trait::async_trait;
-use futures::stream::Stream;
 
 /// Concrete types implementing this trait are Bluetooth Central devices.
 /// They provide methods for retrieving nearby connections and device info.
@@ -26,14 +23,14 @@ pub trait Adapter: Sized {
     /// Retrieve the system-default Bluetooth adapter.
     async fn default() -> Result<Self, anyhow::Error>;
 
-    /// Scan for nearby devices, returning a `Stream` of futures that can be
-    /// iterated over and polled to retrieve `BleDevice`.
-    //
-    // NOTE: Using Boxed dyn here is silly because in cross-platform code there
-    // should only ever be one concrete type implementing Adapter. Change this
-    // to `impl Stream` once impl trait return types are stabilized in traits.
-    // b/289224233.
-    fn scan_devices(&self) -> Result<Pin<Box<dyn Stream<Item = Self::Device>>>, anyhow::Error>;
+    /// Begin scanning for nearby devices.
+    fn start_scan_devices(&mut self) -> Result<(), anyhow::Error>;
+
+    /// Stop scanning for nearby devices.
+    fn stop_scan_devices(&mut self) -> Result<(), anyhow::Error>;
+
+    /// Poll next discovered device.
+    async fn next_device(&mut self) -> Result<Self::Device, anyhow::Error>;
 }
 
 /// Concrete types implementing this trait represent Bluetooth Peripheral devices.

fastpair/rust/src/bluetooth/unsupported/adapter.rs

@@ -12,10 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::pin::Pin;
-
 use async_trait::async_trait;
-use futures::stream::Stream;
 
 use super::BleDevice;
 use crate::bluetooth::common::Adapter;
@@ -32,11 +29,16 @@ impl Adapter for BleAdapter {
         panic!("Unsupported target platform.");
     }
 
-    fn scan_devices(&self) -> Result<Pin<Box<dyn Stream<Item = Self::Device>>>, anyhow::Error> {
+    fn start_scan_devices(&mut self) -> Result<(), anyhow::Error> {
+        panic!("Unsupported target platform.");
+    }
+
+    fn stop_scan_devices(&mut self) -> Result<(), anyhow::Error> {
+        panic!("Unsupported target platform.");
+    }
+
+    async fn next_device(&mut self) -> Result<Self::Device, anyhow::Error> {
         panic!("Unsupported target platform.");
-        #[allow(unreachable_code)]
-        // Sad satisfying trait bounds github.com/rust-lang/rust/issues/55022.
-        Ok(Box::pin(futures::stream::iter(vec![BleDevice {}])))
     }
 }
 

fastpair/rust/src/bluetooth/windows_ble/adapter.rs

@@ -59,6 +59,11 @@ use crate::bluetooth::{common::Adapter, BleDevice};
 /// Concrete type implementing `Adapter`, used for Windows BLE.
 pub struct BleAdapter {
     inner: BluetoothAdapter,
+    // NOTE: Using Boxed dyn here is silly because only one concrete type ever
+    // used. Change this to `impl Stream` once impl trait return types
+    // stabilized for existential types.
+    // b/289224233.
+    device_stream: Option<Pin<Box<dyn Stream<Item = BleDevice> + Send + Sync>>>,
 }
 
 #[async_trait]
@@ -79,10 +84,13 @@ impl Adapter for BleAdapter {
             ));
         }
 
-        Ok(BleAdapter { inner })
+        Ok(BleAdapter {
+            inner,
+            device_stream: None,
+        })
     }
 
-    fn scan_devices(&self) -> Result<Pin<Box<dyn Stream<Item = Self::Device>>>, anyhow::Error> {
+    fn start_scan_devices(&mut self) -> Result<(), anyhow::Error> {
         let watcher = BluetoothLEAdvertisementWatcher::new()?;
         match watcher.SetScanningMode(BluetoothLEScanningMode::Active) {
             Ok(_) => (),
@@ -106,11 +114,17 @@ impl Adapter for BleAdapter {
         let received_handler = TypedEventHandler::new(
             // Move `weak_sender` into closure.
             move |watcher: &Option<BluetoothLEAdvertisementWatcher>,
-                  event_args: &Option<BluetoothLEAdvertisementReceivedEventArgs>| {
+                  event_args: &Option<
+                BluetoothLEAdvertisementReceivedEventArgs,
+            >| {
                 if watcher.is_some() {
                     if let Some(event_args) = event_args {
                         if let Some(sender) = weak_sender.upgrade() {
-                            match sender.lock().unwrap().try_send(event_args.clone()) {
+                            match sender
+                                .lock()
+                                .unwrap()
+                                .try_send(event_args.clone())
+                            {
                                 Ok(_) => (),
                                 Err(err) => {
                                     error!("Error while handling Received event: {:?}", err)
@@ -148,15 +162,18 @@ impl Adapter for BleAdapter {
         // We apply a FilterMap to map from advertisement packet to a future
         // returning `BleDevice` and filter out undesired connections. We need a
         // pinned box to satisfy trait bounds for `Stream`.
-        Ok(Box::pin(receiver.filter_map(move |event_args| {
-            //  Move `watcher` into `FilterMap` closure. This ensures `watcher`
-            // is only dropped when the stream is closed.
-            let _watcher = &watcher;
+        self.device_stream =
+            Some(Box::pin(receiver.filter_map(move |event_args| {
+                //  Move `watcher` into `FilterMap` closure. This ensures `watcher`
+                // is only dropped when the stream is closed.
+                let _watcher = &watcher;
 
-            // Move `event_args` into async block.
-            async move {
-                match event_args.AdvertisementType().ok()? {
-                    BluetoothLEAdvertisementType::NonConnectableUndirected => None,
+                // Move `event_args` into async block.
+                async move {
+                    match event_args.AdvertisementType().ok()? {
+                    BluetoothLEAdvertisementType::NonConnectableUndirected => {
+                        None
+                    }
                     _ => {
                         let addr = event_args.BluetoothAddress().ok()?;
                         let kind = event_args.BluetoothAddressType().ok()?;
@@ -170,8 +187,30 @@ impl Adapter for BleAdapter {
                         }
                     }
                 }
-            }
-        })))
+                }
+            })));
+
+        Ok(())
+    }
+
+    fn stop_scan_devices(&mut self) -> Result<(), anyhow::Error> {
+        if let Some(_) = &self.device_stream {
+            self.device_stream.take();
+            Ok(())
+        } else {
+            Err(anyhow::anyhow!("Device scanning hasn't started."))
+        }
+    }
+
+    async fn next_device(&mut self) -> Result<BleDevice, anyhow::Error> {
+        if let Some(stream) = &mut self.device_stream {
+            stream
+                .next()
+                .await
+                .ok_or(anyhow::anyhow!("Device returned from stream is None."))
+        } else {
+            Err(anyhow::anyhow!("Device scanning hasn't started."))
+        }
     }
 }
 

fastpair/rust/src/main.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use futures::{executor, StreamExt};
+use futures::executor;
 
 mod bluetooth;
 
@@ -20,10 +20,10 @@ use bluetooth::common::{Adapter, Device};
 
 fn main() -> Result<(), anyhow::Error> {
     let run = async {
-        let adapter = bluetooth::BleAdapter::default().await?;
-        let mut scanner = adapter.scan_devices()?;
+        let mut adapter = bluetooth::BleAdapter::default().await?;
+        adapter.start_scan_devices()?;
 
-        while let Some(device) = scanner.next().await {
+        while let Ok(device) = adapter.next_device().await {
             println!("found {}", device.name()?)
         }