fix(corebluetooth): Stop the CoreBluetooth thread once its adapter and peripherals are dropped

src/corebluetooth/adapter.rs

@@ -11,20 +11,59 @@ use crate::common::adapter_manager::AdapterManager;
 use crate::{Error, Result};
 use async_trait::async_trait;
 use futures::channel::mpsc::{self, Sender};
+use futures::channel::oneshot;
 use futures::sink::SinkExt;
 use futures::stream::{Stream, StreamExt};
 use log::*;
 use objc2_core_bluetooth::CBManagerState;
 use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 use tokio::task;
 
 /// Implementation of [api::Central](crate::api::Central).
 #[derive(Clone, Debug)]
 pub struct Adapter {
     manager: Arc<AdapterManager<Peripheral>>,
     sender: Sender<CoreBluetoothMessage>,
+    keep_alive: KeepAlive,
+}
+
+/// Keeps an adapter's CoreBluetooth thread running. The adapter, every
+/// peripheral it hands out and the event and notification streams from them
+/// hold one; the thread stops when the last is dropped.
+/// The adapter's own bookkeeping (its event task and peripheral map) doesn't
+/// hold one, since it lives as long as the thread does.
+pub(crate) type KeepAlive = Arc<oneshot::Sender<()>>;
+
+/// Wraps a stream handed to the application so that it holds a `KeepAlive`,
+/// and keeps delivering items after the adapter it came from is dropped.
+pub(crate) fn keep_alive_stream<T: 'static>(
+    stream: Pin<Box<dyn Stream<Item = T> + Send>>,
+    keep_alive: Option<KeepAlive>,
+) -> Pin<Box<dyn Stream<Item = T> + Send>> {
+    Box::pin(KeepAliveStream {
+        stream,
+        _keep_alive: keep_alive,
+    })
+}
+
+struct KeepAliveStream<T> {
+    stream: Pin<Box<dyn Stream<Item = T> + Send>>,
+    _keep_alive: Option<KeepAlive>,
+}
+
+impl<T> Stream for KeepAliveStream<T> {
+    type Item = T;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+        self.stream.as_mut().poll_next(cx)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 fn get_central_state(state: CBManagerState) -> CentralState {
@@ -38,7 +77,8 @@ fn get_central_state(state: CBManagerState) -> CentralState {
 impl Adapter {
     pub(crate) async fn new() -> Result<Self> {
         let (sender, mut receiver) = mpsc::channel(256);
-        let adapter_sender = run_corebluetooth_thread(sender)?;
+        let (shutdown_sender, shutdown_receiver) = oneshot::channel();
+        let adapter_sender = run_corebluetooth_thread(sender, shutdown_receiver)?;
         // Since init currently blocked until the state update, we know the
         // receiver is dropped after that. We can pick it up here and make it
         // part of our event loop to update our peripherals.
@@ -153,6 +193,7 @@ impl Adapter {
         Ok(Adapter {
             manager,
             sender: adapter_sender,
+            keep_alive: Arc::new(shutdown_sender),
         })
     }
 }
@@ -162,7 +203,10 @@ impl Central for Adapter {
     type Peripheral = Peripheral;
 
     async fn events(&self) -> Result<Pin<Box<dyn Stream<Item = CentralEvent> + Send>>> {
-        Ok(self.manager.event_stream())
+        Ok(keep_alive_stream(
+            self.manager.event_stream(),
+            Some(self.keep_alive.clone()),
+        ))
     }
 
     async fn start_scan(&self, filter: ScanFilter) -> Result<()> {
@@ -182,7 +226,12 @@ impl Central for Adapter {
     }
 
     async fn peripherals(&self) -> Result<Vec<Peripheral>> {
-        Ok(self.manager.peripherals())
+        Ok(self
+            .manager
+            .peripherals()
+            .into_iter()
+            .map(|p| p.with_keep_alive(&self.keep_alive))
+            .collect())
     }
 
     async fn retrieve_peripherals(
@@ -206,7 +255,10 @@ impl Central for Adapter {
             })
             .await?;
         match fut.await {
-            CoreBluetoothReply::Peripherals(peripherals) => Ok(peripherals),
+            CoreBluetoothReply::Peripherals(peripherals) => Ok(peripherals
+                .into_iter()
+                .map(|p| p.with_keep_alive(&self.keep_alive))
+                .collect()),
             CoreBluetoothReply::Err(msg) => Err(Error::RuntimeError(msg)),
             CoreBluetoothReply::Ok => Ok(Vec::new()),
             _ => Err(Error::RuntimeError(
@@ -216,7 +268,10 @@ impl Central for Adapter {
     }
 
     async fn peripheral(&self, id: &PeripheralId) -> Result<Peripheral> {
-        self.manager.peripheral(id).ok_or(Error::DeviceNotFound)
+        self.manager
+            .peripheral(id)
+            .map(|p| p.with_keep_alive(&self.keep_alive))
+            .ok_or(Error::DeviceNotFound)
     }
 
     async fn add_peripheral(&self, _address: &PeripheralId) -> Result<Peripheral> {

src/corebluetooth/internal.rs

@@ -24,6 +24,8 @@ use crate::api::{
     WriteType,
 };
 use futures::channel::mpsc::{self, Receiver, Sender};
+use futures::channel::oneshot;
+use futures::future::{self, Either};
 use futures::select;
 use futures::sink::SinkExt;
 use futures::stream::{Fuse, StreamExt};
@@ -42,6 +44,7 @@ use std::{
     fmt::{self, Debug, Formatter},
     future::Future,
     ops::Deref,
+    pin::pin,
     thread,
 };
 use tokio::runtime;
@@ -2013,8 +2016,11 @@ async fn within_autorelease_pool<F: Future>(future: F) -> F::Output {
     future.await
 }
 
+/// Starts the thread that owns the `CBCentralManager`. It runs until `shutdown`
+/// completes, which happens when the adapter's last `KeepAlive` is dropped.
 pub fn run_corebluetooth_thread(
     event_sender: Sender<CoreBluetoothEvent>,
+    mut shutdown: oneshot::Receiver<()>,
 ) -> Result<Sender<CoreBluetoothMessage>, Error> {
     let authorization = unsafe { CBManager::authorization_class() };
     if authorization != CBManagerAuthorization::AllowedAlways
@@ -2031,9 +2037,18 @@ pub fn run_corebluetooth_thread(
         let runtime = runtime::Builder::new_current_thread().build().unwrap();
         runtime.block_on(async move {
             let mut cbi = CoreBluetoothInternal::new(receiver, event_sender);
+            // The adapter's event task and peripheral map hold message senders
+            // for as long as this thread runs, so the message channel never
+            // closes on its own. Stop when the last `KeepAlive` is dropped.
             loop {
-                within_autorelease_pool(cbi.wait_for_message()).await;
+                let message = pin!(within_autorelease_pool(cbi.wait_for_message()));
+                // Shutdown first, so a steady stream of delegate events can't
+                // keep the thread running.
+                if let Either::Left(_) = future::select(&mut shutdown, message).await {
+                    break;
+                }
             }
+            trace!("Adapter and its peripherals dropped, stopping the CoreBluetooth thread");
         })
     });
     Ok(sender)

src/corebluetooth/peripheral.rs

@@ -5,6 +5,7 @@
 // Licensed under the BSD 3-Clause license. See LICENSE file in the project root
 // for full license information.
 
+use super::adapter::{KeepAlive, keep_alive_stream};
 use super::internal::{
     CoreBluetoothMessage, CoreBluetoothReply, CoreBluetoothReplyFuture, PeripheralEventInternal,
 };
@@ -55,6 +56,9 @@ impl Display for PeripheralId {
 #[derive(Clone)]
 pub struct Peripheral {
     shared: Arc<Shared>,
+    // Set on the copies the adapter hands out, so the CoreBluetooth thread
+    // keeps running while the application still holds this peripheral.
+    keep_alive: Option<KeepAlive>,
 }
 
 struct Shared {
@@ -194,7 +198,15 @@ impl Peripheral {
                 }
             }
         });
-        Self { shared }
+        Self {
+            shared,
+            keep_alive: None,
+        }
+    }
+
+    pub(super) fn with_keep_alive(mut self, keep_alive: &KeepAlive) -> Self {
+        self.keep_alive = Some(keep_alive.clone());
+        self
     }
 
     pub(super) fn update_name(
@@ -599,7 +611,10 @@ impl api::Peripheral for Peripheral {
 
     async fn notifications(&self) -> Result<Pin<Box<dyn Stream<Item = ValueNotification> + Send>>> {
         let receiver = self.shared.notifications_channel.subscribe();
-        Ok(notifications_stream_from_broadcast_receiver(receiver))
+        Ok(keep_alive_stream(
+            notifications_stream_from_broadcast_receiver(receiver),
+            self.keep_alive.clone(),
+        ))
     }
 
     async fn write_descriptor(&self, descriptor: &Descriptor, data: &[u8]) -> Result<()> {