diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a88e7b..4cf3847 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,52 @@ All notable changes to `blew` are documented here. Format follows ## [Unreleased] +### Added + +- `CentralConfig::connect_timeout: Option` — deadline applied to + `Central::connect()`. `Default` sets it to 15s; `None` restores pre-0.3 + unbounded-wait behavior. Applied uniformly across Apple, Linux, and Android + backends; emits `CentralEvent::DeviceDisconnected { cause: Timeout }` when + it fires. +- `BlewError::ConnectTimedOut(DeviceId)` — returned from `Central::connect` + when the new deadline elapses. Distinct from `BlewError::Timeout`, which + remains reserved for adapter-readiness waits (`wait_ready` / `wait_powered`). +- `BlewError::ConnectInFlight(DeviceId)` — returned when `Central::connect` + is called for a device that already has a connect in flight. Previously the + second caller silently stole the first caller's completion. + +### Changed + +- **Android `Central::disconnect` now awaits the `onConnectionStateChange` + callback** (with a 2-second fallback) instead of returning `Ok` immediately + after dispatching the JNI call. This prevents zombie `BluetoothGatt` + handles when the callback never arrives — the fallback path calls a new + Kotlin `forceClose(addr)` that flushes the service cache (`refresh()`) and + closes the GATT handle synchronously, freeing the client-IF slot. +- **Android post-133 cleanup.** `onConnectionStateChange(DISCONNECTED, + status=133)` now calls the hidden `BluetoothGatt.refresh()` before + `close()` to flush the client-side service cache. The `connect()` + stale-cleanup path also calls `refresh()` on the old handle and waits + ~300ms before issuing the fresh `connectGatt()`, matching the canonical + Android BLE back-off recommendation. +- Linux `Central::connect` deadline is now config-driven (via + `CentralConfig::connect_timeout`) and returns `BlewError::ConnectTimedOut` + on elapse. Previously it was hardcoded to 30s and returned the generic + `BlewError::Timeout`. + ### Fixed +- `Central::connect()` could hang indefinitely on Android if + `onConnectionStateChange` never fired (e.g. radio busy, adapter thrash, + status-133 zombie). The new `connect_timeout` default now bounds this; + callers that need the old behavior can opt in with + `CentralConfig { connect_timeout: None, .. }`. +- Overlapping `Central::connect()` calls for the same device no longer + silently orphan the first caller's completion channel. Android, Apple: + the second caller receives `BlewError::ConnectInFlight` immediately. +- Apple `Central::connect` silently evicted any pending completion channel + when called twice concurrently for the same device, leaving the first + caller hanging. Now rejected with `ConnectInFlight`. - `tauri-plugin-blew` no longer fails to locate the Android Kotlin sources when consumed from crates.io. Previously `build.rs` resolved `../blew/android` relative to its own manifest, which works in the workspace but points to a @@ -177,6 +221,61 @@ These were `pub use`'d from `util` but are removed. If you reached into them --- +## Upgrade guide — 0.2.x → Unreleased + +**If you were constructing `CentralConfig` as a struct literal**, it gains a +new field: + +```rust +// Before +let config = CentralConfig { restore_identifier: Some("...".into()) }; + +// After — either spread the default or set the new field explicitly +let config = CentralConfig { + restore_identifier: Some("...".into()), + ..CentralConfig::default() +}; +``` + +**If you were matching on `BlewError::Timeout` for a connect failure on +Linux**, the variant narrowed to `BlewError::ConnectTimedOut(DeviceId)`: + +```rust +// Before — Linux connect timeout returned the generic variant +match central.connect(&id).await { + Err(BlewError::Timeout) => /* ... */, + _ => /* ... */, +} + +// After +match central.connect(&id).await { + Err(BlewError::ConnectTimedOut(_)) => /* ... */, + _ => /* ... */, +} +``` + +`BlewError::Timeout` is still used by `Central::wait_ready`, +`Peripheral::wait_ready`, and `wait_powered`. + +**If you relied on unbounded connect waits**, set +`CentralConfig.connect_timeout = None` when constructing the central: + +```rust +let central: Central = Central::with_config(CentralConfig { + connect_timeout: None, + ..CentralConfig::default() +}).await?; +``` + +**If overlapping `connect()` calls were previously "works by luck"**: the +second caller now receives `BlewError::ConnectInFlight(DeviceId)` immediately +on Apple and Android. If you need shared-completion fan-out semantics, +implement them at your layer (e.g. wrap the shared future in +`futures::future::Shared`). Linux still permits concurrent connects via +bluer's own state machine. + +--- + ## [0.1.0] Initial release. diff --git a/crates/blew/android/src/main/java/org/jakebot/blew/BleCentralManager.kt b/crates/blew/android/src/main/java/org/jakebot/blew/BleCentralManager.kt index 25648bb..7b0a5fa 100644 --- a/crates/blew/android/src/main/java/org/jakebot/blew/BleCentralManager.kt +++ b/crates/blew/android/src/main/java/org/jakebot/blew/BleCentralManager.kt @@ -16,6 +16,7 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.delay import kotlinx.coroutines.launch import java.util.UUID import java.util.concurrent.ConcurrentHashMap @@ -309,6 +310,14 @@ object BleCentralManager { noResponseHandled.remove(addr) pendingNonces.keys.removeIf { it.startsWith("$addr:") } gattQueues.remove(addr)?.close(CancellationException("device $addr disconnected")) + // Status 133 is the Android BLE zombie signal. Flush the + // client-side service cache before close() so the next + // connectGatt() on this address starts with a clean slate. + if (status == 133) { + if (refreshGatt(gatt)) { + Log.d(TAG, "flushed GATT cache for $addr after status=133") + } + } gatt.close() nativeOnConnectionStateChanged(addr, false, status) } @@ -413,16 +422,34 @@ object BleCentralManager { } // Close any stale GATT connection to avoid leaking clientIf slots. - // Android has a limit of ~7 concurrent GATT clients. - gattConnections.remove(deviceAddr)?.let { oldGatt -> - oldGatt.disconnect() - oldGatt.close() + // Android has a limit of ~7 concurrent GATT clients. If we had a stale + // handle, flush its service cache and give the stack ~300ms to release + // the client-IF before the next connectGatt — back-to-back attempts on + // the same address can be silently dropped on some vendors. + val stale = gattConnections.remove(deviceAddr) + if (stale != null) { + refreshGatt(stale) + stale.disconnect() + stale.close() Log.d(TAG, "closed stale GATT for $deviceAddr") + scope.launch { + delay(300) + openGatt(ctx, deviceAddr) + } + return } + openGatt(ctx, deviceAddr) + } + + private fun openGatt( + ctx: Context, + deviceAddr: String, + ) { val device = adapter?.getRemoteDevice(deviceAddr) ?: run { Log.e(TAG, "could not get remote device $deviceAddr") + nativeOnConnectionStateChanged(deviceAddr, false, 0) return } // TRANSPORT_LE ensures we connect over BLE, not classic Bluetooth. @@ -443,6 +470,46 @@ object BleCentralManager { } } + /** + * Synchronously tear down the GATT handle for [deviceAddr] without waiting + * for [BluetoothGattCallback.onConnectionStateChange]. Called from Rust + * when the normal disconnect callback path cannot be trusted (connect + * timeout, disconnect whose callback never arrived, status-133 zombie). + * + * Flushes the client-side service cache with `refresh()` before closing + * so the next connectGatt() starts clean. Emits a synthetic + * [nativeOnConnectionStateChanged]`(addr, false, 0)` so any Rust-side + * state waiting on the disconnect callback unblocks. + */ + @JvmStatic + fun forceClose(deviceAddr: String) { + val gatt = gattConnections.remove(deviceAddr) + mtuMap.remove(deviceAddr) + noResponseHandled.remove(deviceAddr) + pendingNonces.keys.removeIf { it.startsWith("$deviceAddr:") } + gattQueues.remove(deviceAddr)?.close(CancellationException("device $deviceAddr force-closed")) + if (gatt != null) { + refreshGatt(gatt) + try { + gatt.disconnect() + } catch (e: Exception) { + Log.w(TAG, "forceClose: disconnect threw for $deviceAddr: ${e.message}") + } + gatt.close() + Log.d(TAG, "forceClose: tore down GATT for $deviceAddr") + } + nativeOnConnectionStateChanged(deviceAddr, false, 0) + } + + private fun refreshGatt(gatt: BluetoothGatt): Boolean = + try { + val method = gatt.javaClass.getMethod("refresh") + method.invoke(gatt) as Boolean + } catch (e: Exception) { + Log.w(TAG, "refresh failed: ${e.message}") + false + } + /** * Clear the GATT service cache for [deviceAddr] by invoking the hidden * `BluetoothGatt.refresh()` method via reflection. Returns false if no @@ -453,13 +520,7 @@ object BleCentralManager { @JvmStatic fun refresh(deviceAddr: String): Boolean { val gatt = gattConnections[deviceAddr] ?: return false - return try { - val method = gatt.javaClass.getMethod("refresh") - method.invoke(gatt) as Boolean - } catch (e: Exception) { - Log.w(TAG, "refresh failed for $deviceAddr: ${e.message}") - false - } + return refreshGatt(gatt) } // ── GATT operations (serialized via per-device queue) ── diff --git a/crates/blew/examples/restore.rs b/crates/blew/examples/restore.rs index 9c4524e..4b67bb3 100644 --- a/crates/blew/examples/restore.rs +++ b/crates/blew/examples/restore.rs @@ -51,6 +51,7 @@ async fn main() -> Result<(), Box> { let central: Central = Central::with_config(CentralConfig { restore_identifier: Some(CENTRAL_RESTORE_ID.into()), + ..CentralConfig::default() }) .await?; diff --git a/crates/blew/src/central/types.rs b/crates/blew/src/central/types.rs index 3269303..3a3f601 100644 --- a/crates/blew/src/central/types.rs +++ b/crates/blew/src/central/types.rs @@ -1,14 +1,34 @@ use crate::types::{BleDevice, DeviceId}; use bytes::Bytes; +use std::time::Duration; use uuid::Uuid; +/// Default deadline for [`Central::connect`](crate::central::Central::connect). +/// Covers ACL link-up, optional SMP re-bond, service discovery, and the +/// per-backend post-connect steps (Android's MTU negotiation in particular). +pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + /// Configuration for initialising the central role. -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct CentralConfig { /// On Apple platforms, passed as `CBCentralManagerOptionRestoreIdentifierKey` to /// `initWithDelegate:queue:options:`, enabling state restoration for the app's /// background BLE central session. Ignored on all other platforms. pub restore_identifier: Option, + /// Deadline applied to [`Central::connect`](crate::central::Central::connect). + /// `None` disables the deadline (pre-0.3 behaviour); `Some(d)` returns + /// [`BlewError::ConnectTimedOut`](crate::error::BlewError::ConnectTimedOut) + /// if the connection has not completed within `d`. + pub connect_timeout: Option, +} + +impl Default for CentralConfig { + fn default() -> Self { + Self { + restore_identifier: None, + connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT), + } + } } /// Reason a peripheral disconnected. Used by [`CentralEvent::DeviceDisconnected`]. diff --git a/crates/blew/src/error.rs b/crates/blew/src/error.rs index 9b49dc8..d0add87 100644 --- a/crates/blew/src/error.rs +++ b/crates/blew/src/error.rs @@ -40,6 +40,12 @@ pub enum BlewError { #[error("operation timed out")] Timeout, + #[error("connect to {0} timed out")] + ConnectTimedOut(DeviceId), + + #[error("connect to {0} already in flight")] + ConnectInFlight(DeviceId), + #[error("GATT busy on device {0} (another operation in flight)")] GattBusy(DeviceId), diff --git a/crates/blew/src/platform/android/central.rs b/crates/blew/src/platform/android/central.rs index 02cf8d6..acb73f2 100644 --- a/crates/blew/src/platform/android/central.rs +++ b/crates/blew/src/platform/android/central.rs @@ -1,36 +1,43 @@ use std::collections::HashMap; use std::future::Future; use std::sync::OnceLock; +use std::time::Duration; use jni::objects::{JObject, JObjectArray}; use jni::{jni_sig, jni_str}; use parking_lot::Mutex; use tokio::sync::{broadcast, oneshot}; -use tracing::debug; +use tracing::{debug, warn}; use uuid::Uuid; use crate::central::backend::{self, CentralBackend}; -use crate::central::types::{CentralConfig, CentralEvent, ScanFilter, WriteType}; +use crate::central::types::{CentralConfig, CentralEvent, DisconnectCause, ScanFilter, WriteType}; use crate::error::{BlewError, BlewResult}; use crate::gatt::props::CharacteristicProperties; use crate::gatt::service::{GattCharacteristic, GattService}; use crate::l2cap::{L2capChannel, types::Psm}; use crate::types::{BleDevice, DeviceId}; use crate::util::BroadcastEventStream; -use crate::util::request_map::RequestMap; +use crate::util::request_map::KeyedRequestMap; use super::jni_globals::{central_class, jvm}; +/// How long [`AndroidCentral::disconnect`] waits for the Kotlin-side +/// `onConnectionStateChange(DISCONNECTED)` callback before force-closing the +/// GATT handle. Short — we only need enough time for a well-behaved stack to +/// dispatch the callback; a dead callback is exactly the case we're guarding +/// against. +const DISCONNECT_CALLBACK_TIMEOUT: Duration = Duration::from_secs(2); + struct CentralState { event_tx: broadcast::Sender, - pending_ops: RequestMap>>>, - pending_connects: RequestMap>>, - connect_keys: Mutex>, - pending_discover: RequestMap>>>, - discover_keys: Mutex>, + pending_ops: KeyedRequestMap>>>, + pending_connects: KeyedRequestMap>>, + pending_disconnects: KeyedRequestMap>, + pending_discover: KeyedRequestMap>>>, discovered: Mutex>, mtu_cache: Mutex>, - pending_op_keys: Mutex>, + connect_timeout: Mutex>, } static STATE: OnceLock = OnceLock::new(); @@ -39,19 +46,28 @@ fn state() -> &'static CentralState { STATE.get().expect("AndroidCentral not initialized") } -fn init_statics() { +fn init_statics(connect_timeout: Option) { let (event_tx, _) = broadcast::channel(256); - let _ = STATE.set(CentralState { - event_tx, - pending_ops: RequestMap::new(), - pending_connects: RequestMap::new(), - connect_keys: Mutex::new(HashMap::new()), - pending_discover: RequestMap::new(), - discover_keys: Mutex::new(HashMap::new()), - discovered: Mutex::new(Vec::new()), - mtu_cache: Mutex::new(HashMap::new()), - pending_op_keys: Mutex::new(HashMap::new()), - }); + let first_init = STATE + .set(CentralState { + event_tx, + pending_ops: KeyedRequestMap::new(), + pending_connects: KeyedRequestMap::new(), + pending_disconnects: KeyedRequestMap::new(), + pending_discover: KeyedRequestMap::new(), + discovered: Mutex::new(Vec::new()), + mtu_cache: Mutex::new(HashMap::new()), + connect_timeout: Mutex::new(connect_timeout), + }) + .is_ok(); + if !first_init { + // OnceLock is already populated (AndroidCentral initialised more than + // once in the process). Update the timeout since the existing + // callbacks/maps are still valid for the new instance. + if let Some(s) = STATE.get() { + *s.connect_timeout.lock() = connect_timeout; + } + } super::l2cap_state::init_statics(); } @@ -73,32 +89,34 @@ pub(crate) fn update_discovered(device: BleDevice) { } pub(crate) fn complete_connect(addr: &str, result: BlewResult<()>) { - if let Some(s) = STATE.get() { - if let Some(id) = s.connect_keys.lock().remove(addr) { - if let Some(tx) = s.pending_connects.take(id) { - let _ = tx.send(result); - } - } + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_connects.take(&addr.to_owned()) + { + let _ = tx.send(result); + } +} + +pub(crate) fn complete_disconnect(addr: &str) { + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_disconnects.take(&addr.to_owned()) + { + let _ = tx.send(()); } } pub(crate) fn complete_discover_services(addr: &str, result: BlewResult>) { - if let Some(s) = STATE.get() { - if let Some(id) = s.discover_keys.lock().remove(addr) { - if let Some(tx) = s.pending_discover.take(id) { - let _ = tx.send(result); - } - } + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_discover.take(&addr.to_owned()) + { + let _ = tx.send(result); } } pub(crate) fn complete_pending(key: &str, result: BlewResult>) { - if let Some(s) = STATE.get() { - if let Some(id) = s.pending_op_keys.lock().remove(key) { - if let Some(tx) = s.pending_ops.take(id) { - let _ = tx.send(result); - } - } + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_ops.take(&key.to_owned()) + { + let _ = tx.send(result); } } @@ -126,10 +144,11 @@ pub(crate) fn parse_services_json(json: &str) -> Vec { .iter() .filter_map(|ch| { let chr_uuid: Uuid = ch.get("uuid")?.as_str()?.parse().ok()?; - let props = ch.get("properties")?.as_i64().unwrap_or(0); + let props = + u16::try_from(ch.get("properties")?.as_i64().unwrap_or(0)).unwrap_or(0); Some(GattCharacteristic { uuid: chr_uuid, - properties: CharacteristicProperties::from_bits_truncate(props as u16), + properties: CharacteristicProperties::from_bits_truncate(props), permissions: crate::gatt::props::AttributePermissions::empty(), value: vec![], descriptors: vec![], @@ -172,8 +191,16 @@ fn gatt_status_to_error(status: i32, device_id: &DeviceId, char_uuid: Uuid) -> B pub struct AndroidCentral; impl AndroidCentral { - pub async fn with_config(_config: CentralConfig) -> crate::error::BlewResult { - Self::new().await + // No awaits in the body (all JNI work is synchronous), but the public API + // mirrors the Apple/Linux async initializers for cross-platform parity. + #[allow(clippy::unused_async)] + pub async fn with_config(config: CentralConfig) -> crate::error::BlewResult { + if !super::are_ble_permissions_granted() { + return Err(BlewError::PermissionDenied); + } + init_statics(config.connect_timeout); + debug!(connect_timeout = ?config.connect_timeout, "AndroidCentral initialized"); + Ok(AndroidCentral) } /// Clear the GATT cache for `device_id` by calling the hidden @@ -199,7 +226,7 @@ impl AndroidCentral { )?; result.z() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if ok { Ok(()) @@ -215,137 +242,160 @@ impl backend::private::Sealed for AndroidCentral {} impl CentralBackend for AndroidCentral { type EventStream = BroadcastEventStream; - fn new() -> impl Future> + Send + async fn new() -> BlewResult where Self: Sized, { - async { - if !super::are_ble_permissions_granted() { - return Err(BlewError::PermissionDenied); - } - init_statics(); - debug!("AndroidCentral initialized"); - Ok(AndroidCentral) - } + Self::with_config(CentralConfig::default()).await } - fn is_powered(&self) -> impl Future> + Send { - async { - jvm() - .attach_current_thread(|env| { - let result = env.call_static_method( - central_class(), - jni_str!("isPowered"), - jni_sig!("()Z"), - &[], - )?; - Ok(result.z()?) - }) - .map_err(jni_err) - } + async fn is_powered(&self) -> BlewResult { + jvm() + .attach_current_thread(|env| { + let result = env.call_static_method( + central_class(), + jni_str!("isPowered"), + jni_sig!("()Z"), + &[], + )?; + result.z() + }) + .map_err(|e| jni_err(&e)) } - fn start_scan(&self, filter: ScanFilter) -> impl Future> + Send { - async move { - let low_power = filter.mode == crate::central::types::ScanMode::LowPower; - jvm() - .attach_current_thread(|env| { - let string_class = env.find_class(jni_str!("java/lang/String"))?; - let uuids: JObjectArray = env.new_object_array( - filter.services.len() as i32, - &string_class, - &JObject::null(), - )?; - for (i, uuid) in filter.services.iter().enumerate() { - let s = env.new_string(uuid.to_string())?; - uuids.set_element(env, i, &s)?; - } + async fn start_scan(&self, filter: ScanFilter) -> BlewResult<()> { + let low_power = filter.mode == crate::central::types::ScanMode::LowPower; + let service_count = i32::try_from(filter.services.len()) + .map_err(|_| BlewError::Internal("scan filter has too many services".into()))?; + jvm() + .attach_current_thread(|env| { + let string_class = env.find_class(jni_str!("java/lang/String"))?; + let uuids: JObjectArray = + env.new_object_array(service_count, &string_class, JObject::null())?; + for (i, uuid) in filter.services.iter().enumerate() { + let s = env.new_string(uuid.to_string())?; + uuids.set_element(env, i, &s)?; + } - env.call_static_method( - central_class(), - jni_str!("startScan"), - jni_sig!("([Ljava/lang/String;Z)V"), - &[(&uuids).into(), low_power.into()], - )?; + env.call_static_method( + central_class(), + jni_str!("startScan"), + jni_sig!("([Ljava/lang/String;Z)V"), + &[(&uuids).into(), low_power.into()], + )?; - Ok(()) - }) - .map_err(jni_err)?; + Ok(()) + }) + .map_err(|e| jni_err(&e))?; - Ok(()) - } + Ok(()) } - fn stop_scan(&self) -> impl Future> + Send { - async { - jvm() - .attach_current_thread(|env| { - env.call_static_method( - central_class(), - jni_str!("stopScan"), - jni_sig!("()V"), - &[], - )?; - Ok(()) - }) - .map_err(jni_err)?; - Ok(()) - } + async fn stop_scan(&self) -> BlewResult<()> { + jvm() + .attach_current_thread(|env| { + env.call_static_method( + central_class(), + jni_str!("stopScan"), + jni_sig!("()V"), + &[], + )?; + Ok(()) + }) + .map_err(|e| jni_err(&e))?; + Ok(()) } - fn discovered_devices(&self) -> impl Future>> + Send { - async { - let list = STATE - .get() - .map(|s| s.discovered.lock().clone()) - .unwrap_or_default(); - Ok(list) - } + async fn discovered_devices(&self) -> BlewResult> { + let list = STATE + .get() + .map(|s| s.discovered.lock().clone()) + .unwrap_or_default(); + Ok(list) } fn connect(&self, device_id: &DeviceId) -> impl Future> + Send { let addr = device_id.as_str().to_owned(); - let id_for_err = device_id.clone(); + let did = device_id.clone(); async move { + let s = state(); let (tx, rx) = oneshot::channel(); - let s = state(); - let id = s.pending_connects.insert(tx); - s.connect_keys.lock().insert(addr.clone(), id); + if s.pending_connects.try_insert(addr.clone(), tx).is_err() { + return Err(BlewError::ConnectInFlight(did)); + } - jvm() - .attach_current_thread(|env| { - let j_addr = env.new_string(&addr)?; - env.call_static_method( - central_class(), - jni_str!("connect"), - jni_sig!("(Ljava/lang/String;)V"), - &[(&j_addr).into()], - )?; - Ok(()) - }) - .map_err(jni_err)?; + if let Err(err) = jvm().attach_current_thread(|env| { + let j_addr = env.new_string(&addr)?; + env.call_static_method( + central_class(), + jni_str!("connect"), + jni_sig!("(Ljava/lang/String;)V"), + &[(&j_addr).into()], + )?; + Ok(()) + }) { + s.pending_connects.take(&addr); + return Err(jni_err(&err)); + } - rx.await - .map_err(|_| BlewError::DisconnectedDuringOperation(id_for_err))? + let timeout = *s.connect_timeout.lock(); + match timeout { + Some(dur) => match tokio::time::timeout(dur, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(BlewError::DisconnectedDuringOperation(did)), + Err(_) => { + s.pending_connects.take(&addr); + force_close_gatt(&addr); + let _ = s.event_tx.send(CentralEvent::DeviceDisconnected { + device_id: did.clone(), + cause: DisconnectCause::Timeout, + }); + Err(BlewError::ConnectTimedOut(did)) + } + }, + None => rx + .await + .map_err(|_| BlewError::DisconnectedDuringOperation(did))?, + } } } fn disconnect(&self, device_id: &DeviceId) -> impl Future> + Send { let addr = device_id.as_str().to_owned(); async move { - jvm() - .attach_current_thread(|env| { - let j_addr = env.new_string(&addr)?; - env.call_static_method( - central_class(), - jni_str!("disconnect"), - jni_sig!("(Ljava/lang/String;)V"), - &[(&j_addr).into()], - )?; - Ok(()) - }) - .map_err(jni_err)?; + let s = state(); + let (tx, rx) = oneshot::channel(); + let awaits_callback = s.pending_disconnects.try_insert(addr.clone(), tx).is_ok(); + + if let Err(err) = jvm().attach_current_thread(|env| { + let j_addr = env.new_string(&addr)?; + env.call_static_method( + central_class(), + jni_str!("disconnect"), + jni_sig!("(Ljava/lang/String;)V"), + &[(&j_addr).into()], + )?; + Ok(()) + }) { + if awaits_callback { + s.pending_disconnects.take(&addr); + } + return Err(jni_err(&err)); + } + + if !awaits_callback { + return Ok(()); + } + + if tokio::time::timeout(DISCONNECT_CALLBACK_TIMEOUT, rx) + .await + .is_err() + { + warn!(device = %addr, "disconnect callback did not fire; force-closing GATT"); + s.pending_disconnects.take(&addr); + force_close_gatt(&addr); + } Ok(()) } } @@ -360,8 +410,9 @@ impl CentralBackend for AndroidCentral { let (tx, rx) = oneshot::channel(); let s = state(); - let id = s.pending_discover.insert(tx); - s.discover_keys.lock().insert(addr.clone(), id); + if let Some(evicted) = s.pending_discover.insert(addr.clone(), tx) { + let _ = evicted.send(Err(BlewError::GattBusy(did.clone()))); + } let status = jvm() .attach_current_thread(|env| { @@ -372,14 +423,12 @@ impl CentralBackend for AndroidCentral { jni_sig!("(Ljava/lang/String;)I"), &[(&j_addr).into()], )?; - Ok(result.i()?) + result.i() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if status != STATUS_SUCCESS { - // Clean up the pending oneshot since the callback won't fire. - s.discover_keys.lock().remove(&addr); - s.pending_discover.take(id); + s.pending_discover.take(&addr); return Err(gatt_status_to_error(status, &did, Uuid::nil())); } @@ -400,8 +449,9 @@ impl CentralBackend for AndroidCentral { let key = format!("{addr}:read:{char_uuid}"); let s = state(); - let id = s.pending_ops.insert(tx); - s.pending_op_keys.lock().insert(key.clone(), id); + if let Some(evicted) = s.pending_ops.insert(key.clone(), tx) { + let _ = evicted.send(Err(BlewError::GattBusy(did.clone()))); + } let status = jvm() .attach_current_thread(|env| { @@ -414,13 +464,12 @@ impl CentralBackend for AndroidCentral { &[(&j_addr).into(), (&j_uuid).into()], )?; - Ok(result.i()?) + result.i() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if status != STATUS_SUCCESS { - s.pending_op_keys.lock().remove(&key); - s.pending_ops.take(id); + s.pending_ops.take(&key); return Err(gatt_status_to_error(status, &did, char_uuid)); } @@ -447,14 +496,15 @@ impl CentralBackend for AndroidCentral { let s = state(); // For write-without-response, don't wait for a callback. - let (rx, pending_key, pending_id) = if write_type == WriteType::WithResponse { + let (rx, pending_key) = if write_type == WriteType::WithResponse { let (tx, rx) = oneshot::channel(); let key = format!("{addr}:write:{char_uuid}"); - let id = s.pending_ops.insert(tx); - s.pending_op_keys.lock().insert(key.clone(), id); - (Some(rx), Some(key), Some(id)) + if let Some(evicted) = s.pending_ops.insert(key.clone(), tx) { + let _ = evicted.send(Err(BlewError::GattBusy(did.clone()))); + } + (Some(rx), Some(key)) } else { - (None, None, None) + (None, None) }; let status = jvm() @@ -475,14 +525,13 @@ impl CentralBackend for AndroidCentral { ], )?; - Ok(result.i()?) + result.i() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if status != STATUS_SUCCESS { - if let (Some(key), Some(id)) = (pending_key, pending_id) { - s.pending_op_keys.lock().remove(&key); - s.pending_ops.take(id); + if let Some(key) = pending_key { + s.pending_ops.take(&key); } return Err(gatt_status_to_error(status, &did, char_uuid)); } @@ -516,9 +565,9 @@ impl CentralBackend for AndroidCentral { &[(&j_addr).into(), (&j_uuid).into()], )?; - Ok(result.i()?) + result.i() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if status != STATUS_SUCCESS { return Err(gatt_status_to_error(status, &did, char_uuid)); @@ -548,9 +597,9 @@ impl CentralBackend for AndroidCentral { &[(&j_addr).into(), (&j_uuid).into()], )?; - Ok(result.i()?) + result.i() }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; if status != STATUS_SUCCESS { return Err(gatt_status_to_error(status, &did, char_uuid)); @@ -588,11 +637,11 @@ impl CentralBackend for AndroidCentral { central_class(), jni_str!("openL2capChannel"), jni_sig!("(Ljava/lang/String;I)V"), - &[(&j_addr).into(), (psm.value() as i32).into()], + &[(&j_addr).into(), i32::from(psm.value()).into()], )?; Ok(()) }) - .map_err(jni_err)?; + .map_err(|e| jni_err(&e))?; rx.await .map_err(|_| BlewError::DisconnectedDuringOperation(id_for_err))? @@ -604,6 +653,27 @@ impl CentralBackend for AndroidCentral { } } -fn jni_err(e: jni::errors::Error) -> BlewError { +fn jni_err(e: &jni::errors::Error) -> BlewError { BlewError::Internal(format!("JNI error: {e}")) } + +/// Synchronously release the Kotlin-side GATT handle for `addr`. Used when +/// the normal disconnect callback path can't be trusted — connect timeout, +/// the status-133 zombie scenario, or a disconnect whose callback never +/// arrives. The Kotlin `forceClose` calls `refresh()` + `gatt.close()` and +/// removes the entry from `gattConnections`, freeing the client-IF slot. +fn force_close_gatt(addr: &str) { + let result: Result<(), jni::errors::Error> = jvm().attach_current_thread(|env| { + let j_addr = env.new_string(addr)?; + env.call_static_method( + central_class(), + jni_str!("forceClose"), + jni_sig!("(Ljava/lang/String;)V"), + &[(&j_addr).into()], + )?; + Ok(()) + }); + if let Err(e) = result { + warn!(device = %addr, error = %e, "force_close_gatt JNI call failed"); + } +} diff --git a/crates/blew/src/platform/android/jni_globals.rs b/crates/blew/src/platform/android/jni_globals.rs index a455a49..060ffa8 100644 --- a/crates/blew/src/platform/android/jni_globals.rs +++ b/crates/blew/src/platform/android/jni_globals.rs @@ -38,8 +38,9 @@ pub fn init_jvm(vm: JavaVM) { let _ = CLASS_CENTRAL.set(central_ref); let _ = CLASS_PERIPHERAL.set(peripheral_ref); - // Don't let JNI try to delete the activity reference -- we don't own it. - std::mem::forget(activity); + // `activity` is a borrowed local ref from ndk_context; `JObject` has no + // Drop impl, so letting it fall out of scope is a no-op. + let _ = activity; Ok::<_, jni::errors::Error>(()) }) diff --git a/crates/blew/src/platform/android/jni_hooks.rs b/crates/blew/src/platform/android/jni_hooks.rs index 554eec8..5b92672 100644 --- a/crates/blew/src/platform/android/jni_hooks.rs +++ b/crates/blew/src/platform/android/jni_hooks.rs @@ -3,6 +3,27 @@ //! Each `#[unsafe(no_mangle)] extern "C"` function matches a Kotlin `external fun` declaration. //! These run on Android Binder threads and must not block -- they push events into //! tokio channels and return immediately. +//! +//! # Safety +//! +//! All functions in this module are `unsafe extern "C"` JNI entry points. They +//! may only be called by the JVM from the Kotlin `external fun` bindings in +//! `BleCentralManager` / `BlePeripheralManager`: `env` must be a valid JNIEnv +//! for the current thread, `JString`/`JByteArray` parameters must be valid +//! JNI handles, and `jint`/`jboolean` parameters are plain integer values. +//! Every function wraps its body in [`guard`] so panics cannot unwind across +//! the FFI boundary. + +// JNI boundary: JNIEnv pointers (`jint`, `jboolean`, `usize → jint` array +// sizes) are fixed-width C types. Narrowing casts on the boundary are +// intentional and well-understood for the values we pass (ATT offsets, +// MTU, PSM, array lengths). +#![allow( + clippy::cast_possible_truncation, + clippy::cast_possible_wrap, + clippy::cast_sign_loss, + clippy::missing_safety_doc +)] use std::panic::{AssertUnwindSafe, catch_unwind}; @@ -68,7 +89,7 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BlePeripheralManager_nativeOnRead let Ok(svc_uuid) = svc.parse::() else { return Ok::<_, jni::errors::Error>(()); }; - let Ok(chr_uuid) = chr.parse::() else { + let Ok(char_id) = chr.parse::() else { return Ok::<_, jni::errors::Error>(()); }; @@ -78,7 +99,7 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BlePeripheralManager_nativeOnRead let event = PeripheralRequest::Read { client_id: DeviceId::from(addr.as_str()), service_uuid: svc_uuid, - char_uuid: chr_uuid, + char_uuid: char_id, offset: offset as u16, responder, }; @@ -154,7 +175,7 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BlePeripheralManager_nativeOnWrit let Ok(svc_uuid) = svc.parse::() else { return Ok::<_, jni::errors::Error>(()); }; - let Ok(chr_uuid) = chr.parse::() else { + let Ok(char_id) = chr.parse::() else { return Ok::<_, jni::errors::Error>(()); }; let data = jbytes_to_vec(env, &value); @@ -176,7 +197,7 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BlePeripheralManager_nativeOnWrit let event = PeripheralRequest::Write { client_id: DeviceId::from(addr.as_str()), service_uuid: svc_uuid, - char_uuid: chr_uuid, + char_uuid: char_id, value: data, responder, }; @@ -224,15 +245,15 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BlePeripheralManager_nativeOnSubs let Some(chr) = jstring_to_string(env, &char_uuid) else { return Ok::<_, jni::errors::Error>(()); }; - let Ok(chr_uuid) = chr.parse::() else { + let Ok(char_id) = chr.parse::() else { return Ok::<_, jni::errors::Error>(()); }; - trace!(addr, %chr_uuid, subscribed = subscribed == JNI_TRUE, "subscription changed"); + trace!(addr, %char_id, subscribed = subscribed == JNI_TRUE, "subscription changed"); super::peripheral::send_state_event(PeripheralStateEvent::SubscriptionChanged { client_id: DeviceId::from(addr.as_str()), - char_uuid: chr_uuid, + char_uuid: char_id, subscribed: subscribed == JNI_TRUE, }); @@ -370,12 +391,14 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BleCentralManager_nativeOnConnect &addr, Err(crate::error::BlewError::NotConnected(device_id.clone())), ); + // Release any caller awaiting disconnect() completion. + super::central::complete_disconnect(&addr); let cause = match gatt_status { - 0 => DisconnectCause::LocalClose, // GATT_SUCCESS + local disconnect - 8 => DisconnectCause::LinkLoss, // GATT_CONN_TIMEOUT + // 0 = GATT_SUCCESS + local disconnect; 22 = GATT_CONN_TERMINATE_LOCAL_HOST. + 0 | 22 => DisconnectCause::LocalClose, + 8 => DisconnectCause::LinkLoss, // GATT_CONN_TIMEOUT 19 => DisconnectCause::RemoteClose, // GATT_CONN_TERMINATE_PEER_USER - 22 => DisconnectCause::LocalClose, // GATT_CONN_TERMINATE_LOCAL_HOST - 133 => DisconnectCause::Gatt133, // the infamous + 133 => DisconnectCause::Gatt133, // the infamous other => DisconnectCause::Unknown(other), }; super::central::send_event(CentralEvent::DeviceDisconnected { @@ -503,14 +526,14 @@ pub unsafe extern "C" fn Java_org_jakebot_blew_BleCentralManager_nativeOnCharact let Some(chr) = jstring_to_string(env, &char_uuid) else { return Ok::<_, jni::errors::Error>(()); }; - let Ok(chr_uuid) = chr.parse::() else { + let Ok(char_id) = chr.parse::() else { return Ok::<_, jni::errors::Error>(()); }; let data = jbytes_to_vec(env, &value); super::central::send_event(CentralEvent::CharacteristicNotification { device_id: DeviceId::from(addr.as_str()), - char_uuid: chr_uuid, + char_uuid: char_id, value: data.into(), }); diff --git a/crates/blew/src/platform/android/l2cap_state.rs b/crates/blew/src/platform/android/l2cap_state.rs index 2f8b530..3d11123 100644 --- a/crates/blew/src/platform/android/l2cap_state.rs +++ b/crates/blew/src/platform/android/l2cap_state.rs @@ -27,10 +27,12 @@ use super::jni_globals::{central_class, jvm, peripheral_class}; const DUPLEX_BUF_SIZE: usize = 65536; const L2CAP_READ_BUF_SIZE: usize = 4096; +type AcceptSender = mpsc::UnboundedSender>; + struct L2capState { pending_server: Mutex>>>, pending_open: Mutex>>>, - accept_tx: Mutex>>>, + accept_tx: Mutex>, data_tx: Mutex>>>, } @@ -60,14 +62,14 @@ pub(crate) fn set_pending_server(tx: oneshot::Sender>) { } pub(crate) fn complete_server_open(result: BlewResult) { - if let Some(s) = STATE.get() { - if let Some(tx) = s.pending_server.lock().take() { - let _ = tx.send(result); - } + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_server.lock().take() + { + let _ = tx.send(result); } } -pub(crate) fn set_accept_tx(tx: mpsc::UnboundedSender>) { +pub(crate) fn set_accept_tx(tx: AcceptSender) { *state().accept_tx.lock() = Some(tx); } @@ -113,7 +115,7 @@ pub(crate) fn on_channel_opened(device_addr: &str, socket_id: i32, from_server: let is_server = from_server; handle.spawn(async move { - let mut buf = vec![0u8; L2CAP_READ_BUF_SIZE]; + let mut buf = vec![0_u8; L2CAP_READ_BUF_SIZE]; loop { match bridge_reader.read(&mut buf).await { Ok(0) | Err(_) => break, @@ -148,29 +150,29 @@ pub(crate) fn on_channel_opened(device_addr: &str, socket_id: i32, from_server: }); if from_server { - if let Some(s) = STATE.get() { - if let Some(tx) = s.accept_tx.lock().as_ref() { - let device_id = DeviceId(device_addr.to_string()); - if tx.send(Ok((device_id, channel))).is_err() { - tracing::warn!( - socket_id, - "L2CAP accept receiver dropped, discarding channel" - ); - } + if let Some(s) = STATE.get() + && let Some(tx) = s.accept_tx.lock().as_ref() + { + let device_id = DeviceId(device_addr.to_string()); + if tx.send(Ok((device_id, channel))).is_err() { + tracing::warn!( + socket_id, + "L2CAP accept receiver dropped, discarding channel" + ); } } - } else if let Some(s) = STATE.get() { - if let Some(tx) = s.pending_open.lock().remove(device_addr) { - let _ = tx.send(Ok(channel)); - } + } else if let Some(s) = STATE.get() + && let Some(tx) = s.pending_open.lock().remove(device_addr) + { + let _ = tx.send(Ok(channel)); } } pub(crate) fn on_channel_data(socket_id: i32, data: &[u8]) { - if let Some(s) = STATE.get() { - if let Some(tx) = s.data_tx.lock().get(&socket_id) { - let _ = tx.send(data.to_vec()); - } + if let Some(s) = STATE.get() + && let Some(tx) = s.data_tx.lock().get(&socket_id) + { + let _ = tx.send(data.to_vec()); } } @@ -181,11 +183,11 @@ pub(crate) fn on_channel_closed(socket_id: i32) { } pub(crate) fn on_channel_error(device_addr: &str, error: String) { - if let Some(s) = STATE.get() { - if let Some(tx) = s.pending_open.lock().remove(device_addr) { - let _ = tx.send(Err(BlewError::L2cap { - source: error.into(), - })); - } + if let Some(s) = STATE.get() + && let Some(tx) = s.pending_open.lock().remove(device_addr) + { + let _ = tx.send(Err(BlewError::L2cap { + source: error.into(), + })); } } diff --git a/crates/blew/src/platform/android/mod.rs b/crates/blew/src/platform/android/mod.rs index df9268a..6cb792f 100644 --- a/crates/blew/src/platform/android/mod.rs +++ b/crates/blew/src/platform/android/mod.rs @@ -11,6 +11,7 @@ pub use peripheral::AndroidPeripheral; /// Check whether the app is running on an Android emulator. /// /// Checks `Build.FINGERPRINT` for known emulator markers ("generic", "emulator", "sdk"). +#[must_use] pub fn is_emulator() -> bool { use jni::{jni_sig, jni_str}; let result: Result = @@ -35,6 +36,7 @@ pub fn is_emulator() -> bool { /// /// Returns `true` when all required permissions are in place, `false` otherwise. /// This is Android-specific; other platforms always return `true`. +#[must_use] pub fn are_ble_permissions_granted() -> bool { use jni::{jni_sig, jni_str}; let result: Result = @@ -45,7 +47,7 @@ pub fn are_ble_permissions_granted() -> bool { jni_sig!("()Z"), &[], )?; - Ok(result.z()?) + result.z() }); result.unwrap_or(false) } @@ -90,7 +92,6 @@ pub fn request_ble_permissions() { jni_sig!("()V"), &[], )?; - drop(activity); Ok(()) }); if let Err(e) = result { diff --git a/crates/blew/src/platform/android/peripheral.rs b/crates/blew/src/platform/android/peripheral.rs index fe741bd..6e5d634 100644 --- a/crates/blew/src/platform/android/peripheral.rs +++ b/crates/blew/src/platform/android/peripheral.rs @@ -1,5 +1,3 @@ -use std::future::Future; - use jni::objects::{JObject, JObjectArray}; use jni::{jni_sig, jni_str}; use parking_lot::Mutex; @@ -55,242 +53,223 @@ impl PeripheralBackend for AndroidPeripheral { type StateEvents = BroadcastEventStream; type Requests = UnboundedReceiverStream; - fn new() -> impl Future> + Send + async fn new() -> BlewResult where Self: Sized, { - async { - if !super::are_ble_permissions_granted() { - return Err(BlewError::PermissionDenied); - } - let (request_tx, request_rx) = mpsc::unbounded_channel(); - let (state_tx, _) = broadcast::channel(256); - *STATE.lock() = Some(PeripheralState { - request_tx, - request_rx: Mutex::new(Some(request_rx)), - state_tx, - }); - debug!("AndroidPeripheral initialized"); - Ok(AndroidPeripheral) + if !super::are_ble_permissions_granted() { + return Err(BlewError::PermissionDenied); } + let (request_tx, request_rx) = mpsc::unbounded_channel(); + let (state_tx, _) = broadcast::channel(256); + *STATE.lock() = Some(PeripheralState { + request_tx, + request_rx: Mutex::new(Some(request_rx)), + state_tx, + }); + debug!("AndroidPeripheral initialized"); + Ok(AndroidPeripheral) } - fn is_powered(&self) -> impl Future> + Send { - async { - jvm() - .attach_current_thread(|env| { - let result = env.call_static_method( - peripheral_class(), - jni_str!("isPowered"), - jni_sig!("()Z"), - &[], - )?; - Ok(result.z()?) - }) - .map_err(jni_err) - } + async fn is_powered(&self) -> BlewResult { + jvm() + .attach_current_thread(|env| { + let result = env.call_static_method( + peripheral_class(), + jni_str!("isPowered"), + jni_sig!("()Z"), + &[], + )?; + result.z() + }) + .map_err(|e| jni_err(&e)) } - fn add_service(&self, service: &GattService) -> impl Future> + Send { + async fn add_service(&self, service: &GattService) -> BlewResult<()> { let service = service.clone(); - async move { - jvm() - .attach_current_thread(|env| { - let service_uuid = env.new_string(service.uuid.to_string())?; - - let n = service.characteristics.len(); - - let string_class = env.find_class(jni_str!("java/lang/String"))?; - let byte_array_class = env.find_class(jni_str!("[B"))?; - - let char_uuids: JObjectArray = - env.new_object_array(n as i32, &string_class, &JObject::null())?; - let char_values: JObjectArray = - env.new_object_array(n as i32, &byte_array_class, &JObject::null())?; - let mut props_arr = vec![0i32; n]; - let mut perms_arr = vec![0i32; n]; - - for (i, ch) in service.characteristics.iter().enumerate() { - let uuid_str = env.new_string(ch.uuid.to_string())?; - char_uuids.set_element(env, i, &uuid_str)?; - - props_arr[i] = blew_props_to_android(ch.properties); - perms_arr[i] = blew_perms_to_android(ch.permissions); - - let value = env.byte_array_from_slice(&ch.value)?; - char_values.set_element(env, i, &value)?; - } - - let j_props = env.new_int_array(n)?; - j_props.set_region(env, 0, &props_arr)?; - - let j_perms = env.new_int_array(n)?; - j_perms.set_region(env, 0, &perms_arr)?; - - env.call_static_method( - peripheral_class(), - jni_str!("addService"), - jni_sig!("(Ljava/lang/String;[Ljava/lang/String;[I[I[[B)V"), - &[ - (&service_uuid).into(), - (&char_uuids).into(), - (&j_props).into(), - (&j_perms).into(), - (&char_values).into(), - ], - )?; + let n = service.characteristics.len(); + let n_i32 = i32::try_from(n) + .map_err(|_| BlewError::Internal("too many characteristics for JNI".into()))?; + jvm() + .attach_current_thread(|env| { + let service_uuid = env.new_string(service.uuid.to_string())?; + + let string_class = env.find_class(jni_str!("java/lang/String"))?; + let byte_array_class = env.find_class(jni_str!("[B"))?; + + let char_uuids: JObjectArray = + env.new_object_array(n_i32, &string_class, JObject::null())?; + let char_values: JObjectArray = + env.new_object_array(n_i32, &byte_array_class, JObject::null())?; + let mut props_arr = vec![0_i32; n]; + let mut perms_arr = vec![0_i32; n]; + + for (i, ch) in service.characteristics.iter().enumerate() { + let uuid_str = env.new_string(ch.uuid.to_string())?; + char_uuids.set_element(env, i, &uuid_str)?; + + props_arr[i] = blew_props_to_android(ch.properties); + perms_arr[i] = blew_perms_to_android(ch.permissions); + + let value = env.byte_array_from_slice(&ch.value)?; + char_values.set_element(env, i, &value)?; + } - Ok(()) - }) - .map_err(jni_err)?; + let j_props = env.new_int_array(n)?; + j_props.set_region(env, 0, &props_arr)?; + + let j_perms = env.new_int_array(n)?; + j_perms.set_region(env, 0, &perms_arr)?; + + env.call_static_method( + peripheral_class(), + jni_str!("addService"), + jni_sig!("(Ljava/lang/String;[Ljava/lang/String;[I[I[[B)V"), + &[ + (&service_uuid).into(), + (&char_uuids).into(), + (&j_props).into(), + (&j_perms).into(), + (&char_values).into(), + ], + )?; + + Ok(()) + }) + .map_err(|e| jni_err(&e))?; - debug!(uuid = %service.uuid, "added GATT service"); - Ok(()) - } + debug!(uuid = %service.uuid, "added GATT service"); + Ok(()) } - fn start_advertising( - &self, - config: &AdvertisingConfig, - ) -> impl Future> + Send { - let config = config.clone(); - async move { - jvm() - .attach_current_thread(|env| { - let name = env.new_string(&config.local_name)?; - - let string_class = env.find_class(jni_str!("java/lang/String"))?; - let uuids: JObjectArray = env.new_object_array( - config.service_uuids.len() as i32, - &string_class, - &JObject::null(), - )?; - for (i, uuid) in config.service_uuids.iter().enumerate() { - let s = env.new_string(uuid.to_string())?; - uuids.set_element(env, i, &s)?; - } + async fn start_advertising(&self, config: &AdvertisingConfig) -> BlewResult<()> { + let uuid_count = i32::try_from(config.service_uuids.len()).map_err(|_| { + BlewError::Internal("too many service UUIDs in AdvertisingConfig".into()) + })?; + jvm() + .attach_current_thread(|env| { + let name = env.new_string(&config.local_name)?; + + let string_class = env.find_class(jni_str!("java/lang/String"))?; + let uuids: JObjectArray = + env.new_object_array(uuid_count, &string_class, JObject::null())?; + for (i, uuid) in config.service_uuids.iter().enumerate() { + let s = env.new_string(uuid.to_string())?; + uuids.set_element(env, i, &s)?; + } - env.call_static_method( - peripheral_class(), - jni_str!("startAdvertising"), - jni_sig!("(Ljava/lang/String;[Ljava/lang/String;)V"), - &[(&name).into(), (&uuids).into()], - )?; + env.call_static_method( + peripheral_class(), + jni_str!("startAdvertising"), + jni_sig!("(Ljava/lang/String;[Ljava/lang/String;)V"), + &[(&name).into(), (&uuids).into()], + )?; - Ok(()) - }) - .map_err(jni_err)?; + Ok(()) + }) + .map_err(|e| jni_err(&e))?; - debug!("advertising started"); - Ok(()) - } + debug!("advertising started"); + Ok(()) } - fn stop_advertising(&self) -> impl Future> + Send { - async { - jvm() - .attach_current_thread(|env| { - env.call_static_method( - peripheral_class(), - jni_str!("stopAdvertising"), - jni_sig!("()V"), - &[], - )?; - Ok(()) - }) - .map_err(jni_err)?; - Ok(()) - } + async fn stop_advertising(&self) -> BlewResult<()> { + jvm() + .attach_current_thread(|env| { + env.call_static_method( + peripheral_class(), + jni_str!("stopAdvertising"), + jni_sig!("()V"), + &[], + )?; + Ok(()) + }) + .map_err(|e| jni_err(&e))?; + Ok(()) } - fn notify_characteristic( + async fn notify_characteristic( &self, device_id: &DeviceId, char_uuid: Uuid, value: Vec, - ) -> impl Future> + Send { + ) -> BlewResult<()> { let device_addr = device_id.as_str().to_owned(); - async move { - // Retry loop: Kotlin returns 1 ("busy") when the previous - // notification hasn't completed yet (onNotificationSent pending). - // We retry with async sleep so we don't block the tokio thread. - for attempt in 0..50u32 { - let status: i32 = jvm() - .attach_current_thread(|env| { - let addr_str = env.new_string(&device_addr)?; - let uuid_str = env.new_string(char_uuid.to_string())?; - let j_value = env.byte_array_from_slice(&value)?; - - let ret = env.call_static_method( - peripheral_class(), - jni_str!("notifyCharacteristic"), - jni_sig!("(Ljava/lang/String;Ljava/lang/String;[B)I"), - &[(&addr_str).into(), (&uuid_str).into(), (&j_value).into()], - )?; - ret.i() - }) - .map_err(jni_err)?; - - match status { - 0 => return Ok(()), // success - 1 => { - // Busy -- previous notification still in flight. - // Yield to tokio and retry after a short delay. - if attempt < 49 { - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - } - } - 2 => return Ok(()), // no subscribers -- not an error - 3 => { - return Err(BlewError::LocalCharacteristicNotFound { char_uuid }); - } - other => { - return Err(BlewError::Peripheral { - source: format!("notify returned unknown status {other}").into(), - }); + // Retry loop: Kotlin returns 1 ("busy") when the previous + // notification hasn't completed yet (onNotificationSent pending). + // We retry with async sleep so we don't block the tokio thread. + for attempt in 0..50_u32 { + let status: i32 = jvm() + .attach_current_thread(|env| { + let addr_str = env.new_string(&device_addr)?; + let uuid_str = env.new_string(char_uuid.to_string())?; + let j_value = env.byte_array_from_slice(&value)?; + + let ret = env.call_static_method( + peripheral_class(), + jni_str!("notifyCharacteristic"), + jni_sig!("(Ljava/lang/String;Ljava/lang/String;[B)I"), + &[(&addr_str).into(), (&uuid_str).into(), (&j_value).into()], + )?; + ret.i() + }) + .map_err(|e| jni_err(&e))?; + + match status { + // 0 = success; 2 = no subscribers (not an error). + 0 | 2 => return Ok(()), + 1 => { + // Busy -- previous notification still in flight. + // Yield to tokio and retry after a short delay. + if attempt < 49 { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; } } + 3 => { + return Err(BlewError::LocalCharacteristicNotFound { char_uuid }); + } + other => { + return Err(BlewError::Peripheral { + source: format!("notify returned unknown status {other}").into(), + }); + } } - // Exhausted retries -- treat as transient error. - Err(BlewError::Peripheral { - source: "notification busy after retries".into(), - }) } + // Exhausted retries -- treat as transient error. + Err(BlewError::Peripheral { + source: "notification busy after retries".into(), + }) } - fn l2cap_listener( + async fn l2cap_listener( &self, - ) -> impl Future< - Output = BlewResult<( - Psm, - impl futures_core::Stream> + Send + 'static, - )>, - > + Send { - async { - let (psm_tx, psm_rx) = oneshot::channel(); - super::l2cap_state::set_pending_server(psm_tx); - - let (accept_tx, accept_rx) = mpsc::unbounded_channel(); - super::l2cap_state::set_accept_tx(accept_tx); - - jvm() - .attach_current_thread(|env| { - env.call_static_method( - peripheral_class(), - jni_str!("openL2capServer"), - jni_sig!("()V"), - &[], - )?; - Ok(()) - }) - .map_err(jni_err)?; + ) -> BlewResult<( + Psm, + impl futures_core::Stream> + Send + 'static, + )> { + let (psm_tx, psm_rx) = oneshot::channel(); + super::l2cap_state::set_pending_server(psm_tx); + + let (accept_tx, accept_rx) = mpsc::unbounded_channel(); + super::l2cap_state::set_accept_tx(accept_tx); + + jvm() + .attach_current_thread(|env| { + env.call_static_method( + peripheral_class(), + jni_str!("openL2capServer"), + jni_sig!("()V"), + &[], + )?; + Ok(()) + }) + .map_err(|e| jni_err(&e))?; - let psm = psm_rx - .await - .map_err(|_| BlewError::Internal("L2CAP server open cancelled".into()))??; + let psm = psm_rx + .await + .map_err(|_| BlewError::Internal("L2CAP server open cancelled".into()))??; - Ok((psm, UnboundedReceiverStream::new(accept_rx))) - } + Ok((psm, UnboundedReceiverStream::new(accept_rx))) } fn state_events(&self) -> Self::StateEvents { @@ -312,13 +291,13 @@ impl PeripheralBackend for AndroidPeripheral { } } -fn jni_err(e: jni::errors::Error) -> BlewError { +fn jni_err(e: &jni::errors::Error) -> BlewError { BlewError::Internal(format!("JNI error: {e}")) } /// Convert blew CharacteristicProperties to Android BluetoothGattCharacteristic property bits. fn blew_props_to_android(props: CharacteristicProperties) -> i32 { - let mut out = 0i32; + let mut out = 0_i32; if props.contains(CharacteristicProperties::BROADCAST) { out |= 0x01; // PROPERTY_BROADCAST } @@ -343,7 +322,7 @@ fn blew_props_to_android(props: CharacteristicProperties) -> i32 { /// Convert blew AttributePermissions to Android permission bits. fn blew_perms_to_android(perms: crate::gatt::props::AttributePermissions) -> i32 { use crate::gatt::props::AttributePermissions; - let mut out = 0i32; + let mut out = 0_i32; if perms.contains(AttributePermissions::READ) { out |= 0x01; // PERMISSION_READ } diff --git a/crates/blew/src/platform/apple/central.rs b/crates/blew/src/platform/apple/central.rs index c080f5d..2f2beea 100644 --- a/crates/blew/src/platform/apple/central.rs +++ b/crates/blew/src/platform/apple/central.rs @@ -92,6 +92,7 @@ struct CentralInner { peripherals: Mutex>>, discovered: Mutex>, connects: KeyedRequestMap>>, + connect_timeout: Mutex>, // `discoveries` keeps a mutable `DiscoveryState` per device (services // accumulate across multiple didDiscoverCharacteristicsForService // callbacks), so it needs `get_mut` and can't use KeyedRequestMap. @@ -121,6 +122,7 @@ impl CentralInner { peripherals: Default::default(), discovered: Default::default(), connects: Default::default(), + connect_timeout: Mutex::new(None), discoveries: Default::default(), reads: Default::default(), writes: Default::default(), @@ -622,6 +624,7 @@ impl CentralBackend for AppleCentral { debug!(device_id = %device_id, "connecting to device"); // All ObjC ops in a synchronous block to avoid holding !Send types across .await. let id_for_err = device_id.clone(); + let peripheral_to_cancel; let rx = { let peripheral = handle .inner @@ -634,19 +637,48 @@ impl CentralBackend for AppleCentral { }; let (tx, rx) = oneshot::channel(); - let evicted = handle.inner.connects.insert(device_id, tx); - if evicted.is_some() { - warn!(device_id = %id_for_err, "concurrent connect evicted pending waiter"); + if handle + .inner + .connects + .try_insert(device_id.clone(), tx) + .is_err() + { + return Err(BlewError::ConnectInFlight(device_id)); } unsafe { peripheral.setDelegate(Some(ProtocolObject::from_ref(&*handle.delegate))); handle.manager.connectPeripheral_options(&peripheral, None); } + peripheral_to_cancel = peripheral; rx }; - rx.await - .unwrap_or(Err(BlewError::DisconnectedDuringOperation(id_for_err))) + + let timeout = *handle.inner.connect_timeout.lock(); + let result = match timeout { + Some(dur) => match tokio::time::timeout(dur, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(BlewError::DisconnectedDuringOperation(id_for_err.clone())), + Err(_) => { + handle.inner.connects.take(&id_for_err); + unsafe { + handle + .manager + .cancelPeripheralConnection(&peripheral_to_cancel); + } + handle.inner.emit(CentralEvent::DeviceDisconnected { + device_id: id_for_err.clone(), + cause: DisconnectCause::Timeout, + }); + Err(BlewError::ConnectTimedOut(id_for_err)) + } + }, + None => rx + .await + .unwrap_or(Err(BlewError::DisconnectedDuringOperation(id_for_err))), + }; + drop(peripheral_to_cancel); + result } } @@ -907,10 +939,9 @@ impl AppleCentral { self.0.inner.restored.lock().take() } - pub async fn with_config( - #[cfg_attr(not(target_os = "ios"), allow(unused))] config: CentralConfig, - ) -> BlewResult { + pub async fn with_config(config: CentralConfig) -> BlewResult { let (inner, mut powered_rx) = CentralInner::new(); + *inner.connect_timeout.lock() = config.connect_timeout; let delegate = CentralDelegate::new(Arc::clone(&inner)); let queue = DispatchQueue::new("blew.central", DispatchQueueAttr::SERIAL); diff --git a/crates/blew/src/platform/linux/central.rs b/crates/blew/src/platform/linux/central.rs index abadef9..1dfc493 100644 --- a/crates/blew/src/platform/linux/central.rs +++ b/crates/blew/src/platform/linux/central.rs @@ -20,8 +20,6 @@ use tokio_stream::StreamExt as _; use tracing::{debug, trace, warn}; use uuid::Uuid; -const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); - struct CentralInner { _session: Session, adapter: Adapter, @@ -31,13 +29,55 @@ struct CentralInner { scan_task: Mutex>>, notify_tasks: Mutex>>, adapter_task: Mutex>>, + connect_timeout: Mutex>, + pending_connects: crate::util::request_map::KeyedRequestMap, } pub struct LinuxCentral(Arc); +async fn connect_inner(handle: Arc, device_id: DeviceId) -> BlewResult<()> { + let addr = CentralInner::parse_addr(&device_id)?; + let device = handle + .adapter + .device(addr) + .map_err(|e| BlewError::Central { + source: Box::new(e), + })?; + let timeout = *handle.connect_timeout.lock(); + let connect_fut = device.connect(); + let result = match timeout { + Some(dur) => tokio::time::timeout(dur, connect_fut).await, + None => Ok(connect_fut.await), + }; + match result { + Ok(Ok(())) => {} + Ok(Err(e)) => { + return Err(BlewError::Central { + source: Box::new(e), + }); + } + Err(_) => { + let _ = device.disconnect().await; + let _ = handle.event_tx.send(CentralEvent::DeviceDisconnected { + device_id: device_id.clone(), + cause: DisconnectCause::Timeout, + }); + return Err(BlewError::ConnectTimedOut(device_id)); + } + } + debug!(device_id = %device_id, "device connected"); + handle.clear_mtu(&device_id); + let _ = handle + .event_tx + .send(CentralEvent::DeviceConnected { device_id }); + Ok(()) +} + impl LinuxCentral { - pub async fn with_config(_config: CentralConfig) -> crate::error::BlewResult { - Self::new().await + pub async fn with_config(config: CentralConfig) -> crate::error::BlewResult { + let this = Self::new().await?; + *this.0.connect_timeout.lock() = config.connect_timeout; + Ok(this) } } @@ -167,6 +207,8 @@ impl CentralBackend for LinuxCentral { let inner = Arc::new(CentralInner { _session: session, adapter, + connect_timeout: Mutex::new(None), + pending_connects: crate::util::request_map::KeyedRequestMap::new(), discovered: Mutex::new(HashMap::new()), mtu_cache: Mutex::new(HashMap::new()), event_tx, @@ -336,28 +378,16 @@ impl CentralBackend for LinuxCentral { let device_id = device_id.clone(); async move { debug!(device_id = %device_id, "connecting to device"); - let addr = CentralInner::parse_addr(&device_id)?; - let device = handle - .adapter - .device(addr) - .map_err(|e| BlewError::Central { - source: Box::new(e), - })?; - match tokio::time::timeout(CONNECT_TIMEOUT, device.connect()).await { - Ok(Ok(())) => {} - Ok(Err(e)) => { - return Err(BlewError::Central { - source: Box::new(e), - }); - } - Err(_) => return Err(BlewError::Timeout), + if handle + .pending_connects + .try_insert(device_id.clone(), ()) + .is_err() + { + return Err(BlewError::ConnectInFlight(device_id)); } - debug!(device_id = %device_id, "device connected"); - handle.clear_mtu(&device_id); - let _ = handle - .event_tx - .send(CentralEvent::DeviceConnected { device_id }); - Ok(()) + let result = connect_inner(Arc::clone(&handle), device_id.clone()).await; + handle.pending_connects.take(&device_id); + result } } diff --git a/crates/blew/src/util/request_map.rs b/crates/blew/src/util/request_map.rs index 913e467..b12ea0a 100644 --- a/crates/blew/src/util/request_map.rs +++ b/crates/blew/src/util/request_map.rs @@ -26,6 +26,21 @@ impl KeyedRequestMap { self.inner.lock().insert(key, value) } + /// Insert `value` at `key` only if the slot is empty. Returns `Ok(())` on + /// success, or `Err(value)` (handing the value back uninserted) if a value + /// was already present. Atomic — no TOCTOU between a caller's `contains` + /// check and `insert`. + pub fn try_insert(&self, key: K, value: V) -> Result<(), V> { + use std::collections::hash_map::Entry; + match self.inner.lock().entry(key) { + Entry::Vacant(slot) => { + slot.insert(value); + Ok(()) + } + Entry::Occupied(_) => Err(value), + } + } + /// Remove and return the value at `key`, if present. pub fn take(&self, key: &K) -> Option { self.inner.lock().remove(key) @@ -98,6 +113,16 @@ mod tests { assert_eq!(map.take(&"a".into()), None); } + #[test] + fn keyed_try_insert_respects_existing() { + let map = KeyedRequestMap::::new(); + assert_eq!(map.try_insert("a".into(), 1), Ok(())); + assert_eq!(map.try_insert("a".into(), 2), Err(2)); + assert_eq!(map.take(&"a".into()), Some(1)); + assert_eq!(map.try_insert("a".into(), 3), Ok(())); + assert_eq!(map.take(&"a".into()), Some(3)); + } + use proptest::collection::vec; use proptest::prelude::*; diff --git a/crates/blew/tests/central_config.rs b/crates/blew/tests/central_config.rs index 413f791..75613cb 100644 --- a/crates/blew/tests/central_config.rs +++ b/crates/blew/tests/central_config.rs @@ -1,5 +1,7 @@ #![cfg(feature = "testing")] +use std::time::Duration; + use blew::{Central, CentralConfig}; #[tokio::test] @@ -7,6 +9,7 @@ use blew::{Central, CentralConfig}; async fn with_config_accepts_restore_identifier() { let config = CentralConfig { restore_identifier: Some("com.example.test".into()), + ..CentralConfig::default() }; let _central = Central::with_config(config).await.unwrap(); } @@ -16,3 +19,18 @@ fn central_config_default_has_no_restore_identifier() { let c = CentralConfig::default(); assert!(c.restore_identifier.is_none()); } + +#[test] +fn central_config_default_sets_15s_connect_timeout() { + let c = CentralConfig::default(); + assert_eq!(c.connect_timeout, Some(Duration::from_secs(15))); +} + +#[test] +fn central_config_connect_timeout_is_opt_outable() { + let c = CentralConfig { + connect_timeout: None, + ..CentralConfig::default() + }; + assert!(c.connect_timeout.is_none()); +}