Merge pull request #10 from akiles/android

Android part 1: scanning

.github/workflows/ci.yml

@@ -51,5 +51,11 @@ jobs:
           rustup target add aarch64-apple-ios
           cargo clippy --all-targets --target=aarch64-apple-ios -- -D warnings
 
+      - name: Clippy (Android)
+        run: |
+          rustup target add aarch64-linux-android
+          # not using --all-targets because examples don't build for Android due to `Adapter::default()`
+          cargo clippy --lib --tests --target=aarch64-linux-android -- -D warnings
+
       - name: Test
         run: cargo test --all

Cargo.toml

@@ -54,6 +54,10 @@ windows = { version = "0.48.0", features = [
 bluer = { version = "0.16.1", features = ["bluetoothd"] }
 tokio = { version = "1.20.1", features = ["rt-multi-thread"] }
 
+[target.'cfg(target_os = "android")'.dependencies]
+java-spaghetti = "0.1.0"
+futures-channel = "0.3.24"
+
 [target.'cfg(any(target_os = "macos", target_os = "ios"))'.dependencies]
 async-broadcast = "0.5.1"
 objc = "0.2.7"

src/adapter.rs

@@ -11,8 +11,24 @@ use crate::{sys, AdapterEvent, AdvertisingDevice, ConnectionEvent, Device, Devic
 pub struct Adapter(sys::adapter::AdapterImpl);
 
 impl Adapter {
+    /// Creates an interface to the default Bluetooth adapter for the system.
+    ///
+    /// # Safety
+    ///
+    /// - `java_vm` must be a valid JNI `JavaVM` pointer to a VM that will stay alive for the entire duration the `Adapter` or any structs obtained from it are live.
+    /// - `bluetooth_manager` must be a valid global reference to an `android.bluetooth.BluetoothManager` instance, from the `java_vm` VM.
+    /// - The `Adapter` takes ownership of the global reference and will delete it with the `DeleteGlobalRef` JNI call when dropped. You must not do that yourself.
+    #[cfg(target_os = "android")]
+    pub unsafe fn new(
+        java_vm: *mut java_spaghetti::sys::JavaVM,
+        bluetooth_manager: java_spaghetti::sys::jobject,
+    ) -> Result<Self> {
+        sys::adapter::AdapterImpl::new(java_vm, bluetooth_manager).map(Self)
+    }
+
     /// Creates an interface to the default Bluetooth adapter for the system
     #[inline]
+    #[cfg(not(target_os = "android"))]
     pub async fn default() -> Option<Self> {
         sys::adapter::AdapterImpl::default().await.map(Adapter)
     }

src/android/adapter.rs

@@ -0,0 +1,316 @@
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicI32, Ordering};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+
+use futures_channel::mpsc::{Receiver, Sender};
+use futures_core::Stream;
+use futures_lite::{stream, StreamExt};
+use java_spaghetti::{Arg, ByteArray, Env, Global, Local, PrimitiveArray, VM};
+use tracing::{debug, warn};
+use uuid::Uuid;
+
+use super::bindings::android::bluetooth::le::{BluetoothLeScanner, ScanResult};
+use super::bindings::android::bluetooth::{BluetoothAdapter, BluetoothManager};
+use super::bindings::android::os::ParcelUuid;
+use super::bindings::com::github::alexmoon::bluest::android::BluestScanCallback;
+use super::device::DeviceImpl;
+use super::{JavaIterator, OptionExt};
+use crate::android::bindings::java::util::Map_Entry;
+use crate::util::defer;
+use crate::{
+    AdapterEvent, AdvertisementData, AdvertisingDevice, ConnectionEvent, Device, DeviceId, ManufacturerData, Result,
+};
+
+struct AdapterInner {
+    manager: Global<BluetoothManager>,
+    _adapter: Global<BluetoothAdapter>,
+    le_scanner: Global<BluetoothLeScanner>,
+}
+
+#[derive(Clone)]
+pub struct AdapterImpl {
+    inner: Arc<AdapterInner>,
+}
+
+impl AdapterImpl {
+    pub unsafe fn new(vm: *mut java_spaghetti::sys::JavaVM, manager: java_spaghetti::sys::jobject) -> Result<Self> {
+        let vm = VM::from_raw(vm);
+        let manager: Global<BluetoothManager> = Global::from_raw(vm, manager);
+
+        vm.with_env(|env| {
+            let local_manager = manager.as_ref(env);
+            let adapter = local_manager.getAdapter()?.non_null()?;
+            let le_scanner = adapter.getBluetoothLeScanner()?.non_null()?;
+
+            Ok(Self {
+                inner: Arc::new(AdapterInner {
+                    _adapter: adapter.as_global(),
+                    le_scanner: le_scanner.as_global(),
+                    manager: manager.clone(),
+                }),
+            })
+        })
+    }
+
+    pub(crate) async fn events(&self) -> Result<impl Stream<Item = Result<AdapterEvent>> + Send + Unpin + '_> {
+        Ok(stream::empty()) // TODO
+    }
+
+    pub async fn wait_available(&self) -> Result<()> {
+        Ok(())
+    }
+
+    pub async fn open_device(&self, _id: &DeviceId) -> Result<Device> {
+        todo!()
+    }
+
+    pub async fn connected_devices(&self) -> Result<Vec<Device>> {
+        todo!()
+    }
+
+    pub async fn connected_devices_with_services(&self, _services: &[Uuid]) -> Result<Vec<Device>> {
+        todo!()
+    }
+
+    pub async fn scan<'a>(
+        &'a self,
+        _services: &'a [Uuid],
+    ) -> Result<impl Stream<Item = AdvertisingDevice> + Send + Unpin + 'a> {
+        self.inner.manager.vm().with_env(|env| {
+            let receiver = SCAN_CALLBACKS.allocate();
+            let callback = BluestScanCallback::new(env, receiver.id)?;
+            let callback_global = callback.as_global();
+            let scanner = self.inner.le_scanner.as_ref(env);
+            scanner.startScan_ScanCallback(&**callback)?;
+
+            let guard = defer(move || {
+                self.inner.manager.vm().with_env(|env| {
+                    let callback = callback_global.as_ref(env);
+                    let scanner = self.inner.le_scanner.as_ref(env);
+                    match scanner.stopScan_ScanCallback(&**callback) {
+                        Ok(()) => debug!("stopped scan"),
+                        Err(e) => warn!("failed to stop scan: {:?}", e),
+                    };
+                });
+            });
+
+            Ok(receiver.map(move |x| {
+                let _guard = &guard;
+                x
+            }))
+        })
+    }
+
+    pub async fn discover_devices<'a>(
+        &'a self,
+        services: &'a [Uuid],
+    ) -> Result<impl Stream<Item = Result<Device>> + Send + Unpin + 'a> {
+        let connected = stream::iter(self.connected_devices_with_services(services).await?).map(Ok);
+
+        // try_unfold is used to ensure we do not start scanning until the connected devices have been consumed
+        let advertising = Box::pin(stream::try_unfold(None, |state| async {
+            let mut stream = match state {
+                Some(stream) => stream,
+                None => self.scan(services).await?,
+            };
+            Ok(stream.next().await.map(|x| (x.device, Some(stream))))
+        }));
+
+        Ok(connected.chain(advertising))
+    }
+
+    pub async fn connect_device(&self, _device: &Device) -> Result<()> {
+        // Windows manages the device connection automatically
+        Ok(())
+    }
+
+    pub async fn disconnect_device(&self, _device: &Device) -> Result<()> {
+        // Windows manages the device connection automatically
+        Ok(())
+    }
+
+    pub async fn device_connection_events<'a>(
+        &'a self,
+        _device: &'a Device,
+    ) -> Result<impl Stream<Item = ConnectionEvent> + Send + Unpin + 'a> {
+        Ok(stream::empty()) // TODO
+    }
+}
+
+impl PartialEq for AdapterImpl {
+    fn eq(&self, _other: &Self) -> bool {
+        true
+    }
+}
+
+impl Eq for AdapterImpl {}
+
+impl std::hash::Hash for AdapterImpl {
+    fn hash<H: std::hash::Hasher>(&self, _state: &mut H) {}
+}
+
+impl std::fmt::Debug for AdapterImpl {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("Adapter").finish()
+    }
+}
+
+static SCAN_CALLBACKS: CallbackRouter<AdvertisingDevice> = CallbackRouter::new();
+
+struct CallbackRouter<T: Send + 'static> {
+    map: Mutex<Option<HashMap<i32, Sender<T>>>>,
+    next_id: AtomicI32,
+}
+
+impl<T: Send + 'static> CallbackRouter<T> {
+    const fn new() -> Self {
+        Self {
+            map: Mutex::new(None),
+            next_id: AtomicI32::new(0),
+        }
+    }
+
+    fn allocate(&'static self) -> CallbackReceiver<T> {
+        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+        let (sender, receiver) = futures_channel::mpsc::channel(16);
+        self.map
+            .lock()
+            .unwrap()
+            .get_or_insert_with(Default::default)
+            .insert(id, sender);
+
+        CallbackReceiver {
+            router: self,
+            id,
+            receiver,
+        }
+    }
+
+    fn callback(&'static self, id: i32, val: T) {
+        if let Some(sender) = self.map.lock().unwrap().as_mut().unwrap().get_mut(&id) {
+            if let Err(e) = sender.try_send(val) {
+                warn!("failed to send scan callback: {:?}", e)
+            }
+        }
+    }
+}
+
+struct CallbackReceiver<T: Send + 'static> {
+    router: &'static CallbackRouter<T>,
+    id: i32,
+    receiver: Receiver<T>,
+}
+
+impl<T: Send + 'static> Stream for CallbackReceiver<T> {
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // safety: this is just a manually-written pin projection.
+        let receiver = unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().receiver) };
+        receiver.poll_next(cx)
+    }
+}
+
+impl<T: Send> Drop for CallbackReceiver<T> {
+    fn drop(&mut self) {
+        self.router.map.lock().unwrap().as_mut().unwrap().remove(&self.id);
+    }
+}
+
+#[no_mangle]
+pub extern "system" fn Java_com_github_alexmoon_bluest_android_BluestScanCallback_nativeOnScanResult(
+    env: Env<'_>,
+    _class: *mut (), // self class, ignore
+    id: i32,
+    callback_type: i32,
+    scan_result: Arg<ScanResult>,
+) {
+    if let Err(e) = on_scan_result(env, id, callback_type, scan_result) {
+        warn!("on_scan_result failed: {:?}", e);
+    }
+}
+
+fn convert_uuid(uuid: Local<'_, ParcelUuid>) -> Result<Uuid> {
+    let uuid = uuid.getUuid()?.non_null()?;
+    let lsb = uuid.getLeastSignificantBits()? as u64;
+    let msb = uuid.getMostSignificantBits()? as u64;
+    Ok(Uuid::from_u64_pair(msb, lsb))
+}
+
+#[no_mangle]
+fn on_scan_result(env: Env<'_>, id: i32, callback_type: i32, scan_result: Arg<ScanResult>) -> Result<()> {
+    let scan_result = unsafe { scan_result.into_ref(env) }.non_null()?;
+
+    tracing::info!("got callback! {} {}", id, callback_type);
+
+    let scan_record = scan_result.getScanRecord()?.non_null()?;
+    let device = scan_result.getDevice()?.non_null()?;
+
+    let address = device.getAddress()?.non_null()?.to_string_lossy();
+    let rssi = scan_result.getRssi()?;
+    let is_connectable = scan_result.isConnectable()?;
+    let local_name = scan_record.getDeviceName()?.map(|s| s.to_string_lossy());
+    let tx_power_level = scan_record.getTxPowerLevel()?;
+
+    // Services
+    let mut services = Vec::new();
+    if let Some(uuids) = scan_record.getServiceUuids()? {
+        for uuid in JavaIterator(uuids.iterator()?.non_null()?) {
+            services.push(convert_uuid(uuid.cast()?)?)
+        }
+    }
+
+    // Service data
+    let mut service_data = HashMap::new();
+    let sd = scan_record.getServiceData()?.non_null()?;
+    let sd = sd.entrySet()?.non_null()?;
+    for entry in JavaIterator(sd.iterator()?.non_null()?) {
+        let entry: Local<Map_Entry> = entry.cast()?;
+        let key: Local<ParcelUuid> = entry.getKey()?.non_null()?.cast()?;
+        let val: Local<ByteArray> = entry.getValue()?.non_null()?.cast()?;
+        service_data.insert(convert_uuid(key)?, val.as_vec().into_iter().map(|i| i as u8).collect());
+    }
+
+    // Manufacturer data
+    let mut manufacturer_data = None;
+    let msd = scan_record.getManufacturerSpecificData()?.non_null()?;
+    // TODO there can be multiple manufacturer data entries, but the bluest API only supports one. So grab just the first.
+    if msd.size()? != 0 {
+        let val: Local<'_, ByteArray> = msd.valueAt(0)?.non_null()?.cast()?;
+        manufacturer_data = Some(ManufacturerData {
+            company_id: msd.keyAt(0)? as _,
+            data: val.as_vec().into_iter().map(|i| i as u8).collect(),
+        });
+    }
+
+    let device_id = DeviceId(address);
+
+    let d = AdvertisingDevice {
+        device: Device(DeviceImpl { id: device_id }),
+        adv_data: AdvertisementData {
+            is_connectable,
+            local_name,
+            manufacturer_data, // TODO, SparseArray is cursed.
+            service_data,
+            services,
+            tx_power_level: Some(tx_power_level as _),
+        },
+        rssi: Some(rssi as _),
+    };
+    SCAN_CALLBACKS.callback(id, d);
+
+    Ok(())
+}
+
+#[no_mangle]
+pub extern "system" fn Java_com_github_alexmoon_bluest_android_BluestScanCallback_nativeOnScanFailed(
+    _env: Env<'_>,
+    _class: *mut (), // self class, ignore
+    id: i32,
+    error_code: i32,
+) {
+    tracing::error!("got scan fail! {} {}", id, error_code);
+    todo!()
+}