diff --git a/src/corebluetooth/adapter.rs b/src/corebluetooth/adapter.rs index b0eb21b2..afb9ffbb 100644 --- a/src/corebluetooth/adapter.rs +++ b/src/corebluetooth/adapter.rs @@ -11,6 +11,7 @@ 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::*; @@ -18,6 +19,7 @@ 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). @@ -25,6 +27,43 @@ use tokio::task; pub struct Adapter { manager: Arc>, sender: Sender, + 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>; + +/// 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( + stream: Pin + Send>>, + keep_alive: Option, +) -> Pin + Send>> { + Box::pin(KeepAliveStream { + stream, + _keep_alive: keep_alive, + }) +} + +struct KeepAliveStream { + stream: Pin + Send>>, + _keep_alive: Option, +} + +impl Stream for KeepAliveStream { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.stream.as_mut().poll_next(cx) + } + + fn size_hint(&self) -> (usize, Option) { + 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 { 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 + 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> { - 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 { - 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 { diff --git a/src/corebluetooth/internal.rs b/src/corebluetooth/internal.rs index 5c03c8c2..1db0e738 100644 --- a/src/corebluetooth/internal.rs +++ b/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(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, + mut shutdown: oneshot::Receiver<()>, ) -> Result, 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) diff --git a/src/corebluetooth/peripheral.rs b/src/corebluetooth/peripheral.rs index 6c1723c0..9e8acd34 100644 --- a/src/corebluetooth/peripheral.rs +++ b/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, + // Set on the copies the adapter hands out, so the CoreBluetooth thread + // keeps running while the application still holds this peripheral. + keep_alive: Option, } 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 + 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<()> {