use anyhow::{format_err, Context as _, Error};
use fidl::endpoints;
use fidl_fuchsia_bluetooth_gatt2::{
Characteristic, CharacteristicNotifierMarker, CharacteristicNotifierRequest,
CharacteristicNotifierRequestStream, ClientEventStream, ClientProxy, Handle, LongReadOptions,
ReadOptions, RemoteServiceEventStream, RemoteServiceProxy, ServiceHandle, ServiceInfo,
ShortReadOptions, WriteMode, WriteOptions,
};
use fidl_fuchsia_bluetooth_le::{
CentralMarker, CentralProxy, ConnectionEventStream, ConnectionOptions, ConnectionProxy, Filter,
ScanOptions, ScanResultWatcherMarker, ScanResultWatcherProxy,
};
use fuchsia_sync::RwLock;
use futures::{select, StreamExt};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use tracing::*;
use {fuchsia_async as fasync, fuchsia_component as app};
use fuchsia_bluetooth::types::le::Peer;
use fuchsia_bluetooth::types::{PeerId, Uuid};
use crate::bluetooth::types::{BleScanResponse, SerializableReadByTypeResult};
use crate::common_utils::common::macros::with_line;
#[derive(Debug)]
struct Client {
proxy: ClientProxy,
_connection: ConnectionProxy,
services: HashMap<u64, ServiceInfo>,
watch_services_task: Option<fasync::Task<()>>,
_events_task: fasync::Task<()>,
}
#[derive(Debug)]
struct Central {
proxy: CentralProxy,
_event_task: fasync::Task<Result<(), Error>>,
}
#[derive(Debug)]
struct RemoteService {
proxy: RemoteServiceProxy,
_event_task: fasync::Task<()>,
notifier_tasks: HashMap<u64, fasync::Task<()>>,
peer_id: PeerId,
service_id: u64,
}
#[derive(Debug)]
pub struct InnerGattClientFacade {
active_remote_service: Option<RemoteService>,
central: Option<Central>,
scan_results: HashMap<PeerId, Peer>,
clients: HashMap<PeerId, Client>,
scan_task: Option<fasync::Task<()>>,
}
#[derive(Debug)]
pub struct GattClientFacade {
inner: Arc<RwLock<InnerGattClientFacade>>,
}
impl GattClientFacade {
pub fn new() -> GattClientFacade {
GattClientFacade {
inner: Arc::new(RwLock::new(InnerGattClientFacade {
active_remote_service: None,
central: None,
scan_results: HashMap::new(),
clients: HashMap::new(),
scan_task: None,
})),
}
}
pub async fn stop_scan(&self) -> Result<(), Error> {
let tag = "GattClientFacade::stop_scan";
if self.inner.write().scan_task.take().is_some() {
info!(tag = &with_line!(tag), "Scan stopped");
} else {
info!(tag = &with_line!(tag), "No scan was running");
}
Ok(())
}
pub async fn start_scan(&self, filter: Option<Filter>) -> Result<(), Error> {
let tag = "GattClientFacade::start_scan";
self.inner.write().scan_results.clear();
GattClientFacade::set_central_proxy(self.inner.clone());
let central = self
.inner
.read()
.central
.as_ref()
.ok_or_else(|| format_err!("No central proxy created."))?
.proxy
.clone();
let options = ScanOptions {
filters: Some(vec![filter.unwrap_or(Filter::default())]),
..Default::default()
};
let (watcher_proxy, watcher_server) =
fidl::endpoints::create_proxy::<ScanResultWatcherMarker>();
let scan_fut = central.scan(&options, watcher_server);
fasync::Task::spawn(async move {
if let Err(e) = scan_fut.await {
warn!(tag = &with_line!(tag), "FIDL error during scan: {:?}", e);
}
})
.detach();
self.inner.write().scan_task = Some(fasync::Task::spawn(
GattClientFacade::scan_result_watcher_task(self.inner.clone(), watcher_proxy),
));
info!(tag = &with_line!(tag), "Scan started");
Ok(())
}
async fn scan_result_watcher_task(
inner: Arc<RwLock<InnerGattClientFacade>>,
watcher_proxy: ScanResultWatcherProxy,
) {
let tag = "GattClientFacade::scan_result_watcher_task";
let mut event_stream = watcher_proxy.take_event_stream();
let mut watch_fut = watcher_proxy.watch();
loop {
select! {
watch_result = watch_fut => {
let peers = match watch_result {
Ok(peers) => peers,
Err(e) => {
info!(
tag = &with_line!(tag),
"FIDL error calling ScanResultWatcher::Watch(): {}", e
);
break;
}};
for fidl_peer in peers {
let peer: Peer = fidl_peer.try_into().unwrap();
debug!(tag = &with_line!(tag), "Peer discovered (id: {}, name: {:?})", peer.id, peer.name);
inner.write().scan_results.insert(peer.id, peer);
}
watch_fut = watcher_proxy.watch();
},
event = event_stream.next() => {
if let Some(Err(err)) = event {
info!(tag = &with_line!(tag), "ScanResultWatcher error: {:?}", err);
}
break; }
}
}
inner.write().scan_task = None;
info!(tag = &with_line!(tag), "ScanResultWatcher closed");
}
async fn active_remote_service_event_task(
inner: Arc<RwLock<InnerGattClientFacade>>,
event_stream: RemoteServiceEventStream,
) {
let tag = "GattClientFacade::active_remote_service_event_task";
event_stream.map(|_| ()).collect::<()>().await;
info!(tag = &with_line!(tag), "RemoteService closed");
inner.write().active_remote_service = None;
}
pub async fn gattc_connect_to_service(
&self,
peer_id: String,
service_id: u64,
) -> Result<(), Error> {
let tag = "GattClientFacade::gattc_connect_to_service";
let peer_id = PeerId::from_str(&peer_id)?;
if let Some(service) = self.inner.read().active_remote_service.as_ref() {
if service.peer_id == peer_id && service.service_id == service_id {
info!(
tag = &with_line!(tag),
"Aready connected to service (peer: {}, service: {})", peer_id, service_id
);
return Ok(());
}
}
self.inner.write().active_remote_service = None;
let client_proxy = self.get_client_proxy(peer_id).ok_or_else(|| {
error!(
tag = &with_line!(tag),
"Unable to connect to service {} (not connected to peer {})", service_id, peer_id
);
format_err!("Not connected to peer")
})?;
let (proxy, server) = endpoints::create_proxy();
client_proxy.connect_to_service(&ServiceHandle { value: service_id }, server)?;
let event_stream = proxy.take_event_stream();
let event_task = fasync::Task::spawn(GattClientFacade::active_remote_service_event_task(
self.inner.clone(),
event_stream,
));
self.inner.write().active_remote_service = Some(RemoteService {
proxy,
_event_task: event_task,
notifier_tasks: HashMap::new(),
peer_id,
service_id,
});
Ok(())
}
pub async fn gattc_discover_characteristics(&self) -> Result<Vec<Characteristic>, Error> {
let discover_characteristics_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("RemoteService proxy not available"))?
.discover_characteristics();
discover_characteristics_fut.await.map_err(|_| format_err!("Failed to send message"))
}
async fn gattc_write_char_internal(
&self,
id: u64,
offset: u16,
write_value: Vec<u8>,
mode: WriteMode,
) -> Result<(), Error> {
let handle = Handle { value: id };
let options =
WriteOptions { offset: Some(offset), write_mode: Some(mode), ..Default::default() };
let write_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("No active service"))?
.write_characteristic(&handle, &write_value, &options);
write_fut
.await
.map_err(|_| format_err!("Failed to send message"))?
.map_err(|err| format_err!("Failed to write characteristic: {:?}", err))
}
pub async fn gattc_write_char_by_id(&self, id: u64, write_value: Vec<u8>) -> Result<(), Error> {
self.gattc_write_char_internal(id, 0, write_value, WriteMode::Default).await
}
pub async fn gattc_write_long_char_by_id(
&self,
id: u64,
offset: u16,
write_value: Vec<u8>,
reliable_mode: bool,
) -> Result<(), Error> {
self.gattc_write_char_internal(
id,
offset,
write_value,
if reliable_mode { WriteMode::Reliable } else { WriteMode::Default },
)
.await
}
pub async fn gattc_write_char_by_id_without_response(
&self,
id: u64,
write_value: Vec<u8>,
) -> Result<(), Error> {
self.gattc_write_char_internal(id, 0, write_value, WriteMode::WithoutResponse).await
}
async fn gattc_read_char_internal(
&self,
id: u64,
options: ReadOptions,
) -> Result<Vec<u8>, Error> {
let handle = Handle { value: id };
let read_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("RemoteService proxy not available"))?
.read_characteristic(&handle, &options);
let read_value = read_fut
.await
.map_err(|_| format_err!("Failed to send message"))?
.map_err(|err| format_err!("Failed to read long characteristic: {:?}", err))?;
Ok(read_value.value.unwrap())
}
pub async fn gattc_read_char_by_id(&self, id: u64) -> Result<Vec<u8>, Error> {
self.gattc_read_char_internal(id, ReadOptions::ShortRead(ShortReadOptions {})).await
}
pub async fn gattc_read_long_char_by_id(
&self,
id: u64,
offset: u16,
max_bytes: u16,
) -> Result<Vec<u8>, Error> {
self.gattc_read_char_internal(
id,
ReadOptions::LongRead(LongReadOptions {
offset: Some(offset),
max_bytes: Some(max_bytes),
..Default::default()
}),
)
.await
}
pub async fn gattc_read_char_by_type(
&self,
raw_uuid: String,
) -> Result<Vec<SerializableReadByTypeResult>, Error> {
let uuid = Uuid::from_str(&raw_uuid)
.map_err(|e| format_err!("Unable to convert to Uuid: {:?}", e))?;
let fidl_uuid = fidl_fuchsia_bluetooth::Uuid::from(uuid);
let read_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("RemoteService proxy not available"))?
.read_by_type(&fidl_uuid);
let results = read_fut
.await
.map_err(|err| format_err!("FIDL error: {:?}", err))?
.map_err(|err| format_err!("Failed to read characteristic by type: {:?}", err))?
.into_iter()
.filter(|r| r.error.is_none())
.map(|r| SerializableReadByTypeResult::new(r).unwrap())
.collect();
Ok(results)
}
async fn gattc_read_desc_internal(
&self,
id: u64,
options: ReadOptions,
) -> Result<Vec<u8>, Error> {
let handle = Handle { value: id };
let read_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("RemoteService proxy not available"))?
.read_descriptor(&handle, &options);
let read_value = read_fut
.await
.map_err(|_| format_err!("Failed to send message"))?
.map_err(|err| format_err!("Failed to read descriptor: {:?}", err))?;
Ok(read_value.value.unwrap())
}
pub async fn gattc_read_desc_by_id(&self, id: u64) -> Result<Vec<u8>, Error> {
self.gattc_read_desc_internal(id, ReadOptions::ShortRead(ShortReadOptions {})).await
}
pub async fn gattc_read_long_desc_by_id(
&self,
id: u64,
offset: u16,
max_bytes: u16,
) -> Result<Vec<u8>, Error> {
self.gattc_read_desc_internal(
id,
ReadOptions::LongRead(LongReadOptions {
offset: Some(offset),
max_bytes: Some(max_bytes),
..Default::default()
}),
)
.await
}
pub async fn gattc_write_desc_by_id(&self, id: u64, write_value: Vec<u8>) -> Result<(), Error> {
self.gattc_write_long_desc_by_id(id, 0, write_value).await
}
pub async fn gattc_write_long_desc_by_id(
&self,
id: u64,
offset: u16,
write_value: Vec<u8>,
) -> Result<(), Error> {
let handle = Handle { value: id };
let options = WriteOptions { offset: Some(offset), ..Default::default() };
let write_fut = self
.get_remote_service_proxy()
.ok_or_else(|| format_err!("RemoteService proxy not available"))?
.write_descriptor(&handle, &write_value, &options);
write_fut
.await
.map_err(|_| format_err!("Failed to send message"))?
.map_err(|err| format_err!("Failed to write descriptor: {:?}", err))
}
async fn notifier_task(
inner: Arc<RwLock<InnerGattClientFacade>>,
id: u64,
mut request_stream: CharacteristicNotifierRequestStream,
) {
let tag = "GattClientFacade::notifier_task";
while let Ok(event) = request_stream.select_next_some().await {
match event {
CharacteristicNotifierRequest::OnNotification { value, responder } => {
info!(
tag = &with_line!(tag),
"Received notification (id: {}, value: {:?})",
id,
value.value.unwrap()
);
let _ = responder.send();
}
}
}
info!(tag = &with_line!(tag), "CharacteristicNotifier closed (id: {})", id);
inner.write().active_remote_service.as_mut().and_then(|s| s.notifier_tasks.remove(&id));
}
pub async fn gattc_toggle_notify_characteristic(
&self,
id: u64,
enable: bool,
) -> Result<(), Error> {
let (register_fut, request_stream) = {
let mut inner = self.inner.write();
let service = inner
.active_remote_service
.as_mut()
.ok_or_else(|| format_err!("Not connected to a service"))?;
if !enable {
service.notifier_tasks.remove(&id);
return Ok(());
}
if service.notifier_tasks.contains_key(&id) {
return Ok(());
}
let (client_end, request_stream) =
fidl::endpoints::create_request_stream::<CharacteristicNotifierMarker>();
let register_fut =
service.proxy.register_characteristic_notifier(&Handle { value: id }, client_end);
(register_fut, request_stream)
};
register_fut
.await
.map_err(|e| format_err!("FIDL error: {:?}", e))?
.map_err(|e| format_err!("Error registering notifier: {:?}", e))?;
let notifier_task = fasync::Task::spawn(GattClientFacade::notifier_task(
self.inner.clone(),
id,
request_stream,
));
self.inner
.write()
.active_remote_service
.as_mut()
.ok_or_else(|| format_err!("Not connected to a service"))?
.notifier_tasks
.insert(id, notifier_task);
Ok(())
}
async fn watch_services_and_update_map(
inner: &Arc<RwLock<InnerGattClientFacade>>,
peer_id: &PeerId,
) -> Result<(), Error> {
let client_proxy = inner
.read()
.clients
.get(peer_id)
.ok_or_else(|| format_err!("Not connected to peer"))?
.proxy
.clone();
let watch_fut = client_proxy.watch_services(&[]);
let (updated, removed) =
watch_fut.await.map_err(|_| format_err!("FIDL error calling WatchServices()"))?;
let mut inner = inner.write();
let services = &mut inner
.clients
.get_mut(peer_id)
.ok_or_else(|| format_err!("Not connected to peer"))?
.services;
for handle in removed {
services.remove(&handle.value);
}
for svc in updated {
services.insert(svc.handle.unwrap().value, svc);
}
Ok(())
}
async fn watch_services_task(inner: Arc<RwLock<InnerGattClientFacade>>, peer_id: PeerId) -> () {
loop {
let tag = "GattClientFacade::watch_services_task";
if let Err(err) =
GattClientFacade::watch_services_and_update_map(&inner, &peer_id).await
{
warn!(tag = &with_line!(tag), "{}", err);
return;
}
}
}
pub async fn list_services(&self, id: String) -> Result<Vec<ServiceInfo>, Error> {
let peer_id = PeerId::from_str(&id).map_err(|_| format_err!("Invalid peer id"))?;
{
let inner = self.inner.read();
let client =
inner.clients.get(&peer_id).ok_or_else(|| format_err!("Not connected to peer"))?;
if client.watch_services_task.is_some() {
return Ok(client.services.iter().map(|(_, svc)| svc.clone()).collect());
}
}
GattClientFacade::watch_services_and_update_map(&self.inner, &peer_id).await?;
let task =
fasync::Task::spawn(GattClientFacade::watch_services_task(self.inner.clone(), peer_id));
let mut inner = self.inner.write();
let client =
inner.clients.get_mut(&peer_id).ok_or_else(|| format_err!("Not connected to peer"))?;
client.watch_services_task = Some(task);
Ok(client.services.iter().map(|(_, svc)| svc.clone()).collect())
}
pub fn get_client_proxy(&self, id: PeerId) -> Option<ClientProxy> {
self.inner.read().clients.get(&id).map(|c| c.proxy.clone())
}
async fn central_event_task(inner: Arc<RwLock<InnerGattClientFacade>>) -> Result<(), Error> {
let tag = "GattClientFacade::central_event_task";
let stream = inner
.write()
.central
.as_ref()
.ok_or_else(|| format_err!("Central not set"))?
.proxy
.take_event_stream();
stream.map(|_| ()).collect::<()>().await;
info!(tag = &with_line!(tag), "Central closed");
inner.write().central.take();
return Ok(());
}
pub fn set_central_proxy(inner: Arc<RwLock<InnerGattClientFacade>>) {
if inner.read().central.is_some() {
return;
}
let proxy = app::client::connect_to_protocol::<CentralMarker>()
.context("Failed to connect to BLE Central service.")
.unwrap();
let event_task = fasync::Task::spawn(GattClientFacade::central_event_task(inner.clone()));
inner.write().central = Some(Central { proxy, _event_task: event_task });
}
async fn connection_event_task(
inner: Arc<RwLock<InnerGattClientFacade>>,
mut connection_stream: ConnectionEventStream,
mut client_stream: ClientEventStream,
peer_id: PeerId,
) {
let tag = "GattClientFacade::connection_event_task";
select! {
_ = connection_stream.next() => info!(tag = &with_line!(tag) , "Connection to {} closed", peer_id),
_ = client_stream.next() => info!(tag = &with_line!(tag), "Client for {} closed", peer_id),
}
inner.write().clients.remove(&peer_id);
}
pub async fn connect_peripheral(&self, id: String) -> Result<(), Error> {
let tag = "GattClientFacade::connect_peripheral";
let peer_id = PeerId::from_str(&id)?;
if self.inner.read().clients.contains_key(&peer_id) {
info!(tag = &with_line!(tag), "Already connected to {}", peer_id);
return Ok(());
}
GattClientFacade::set_central_proxy(self.inner.clone());
let (conn_proxy, conn_server_end) = fidl::endpoints::create_proxy();
let options = ConnectionOptions { bondable_mode: Some(true), ..Default::default() };
self.inner
.read()
.central
.as_ref()
.unwrap()
.proxy
.connect(&peer_id.clone().into(), &options, conn_server_end)
.map_err(|_| format_err!("FIDL error when trying to connect()"))?;
let (client_proxy, client_server_end) = fidl::endpoints::create_proxy();
conn_proxy.request_gatt_client(client_server_end)?;
let events_task = fasync::Task::spawn(GattClientFacade::connection_event_task(
self.inner.clone(),
conn_proxy.take_event_stream(),
client_proxy.take_event_stream(),
peer_id.clone(),
));
self.inner.write().clients.insert(
peer_id,
Client {
proxy: client_proxy,
_connection: conn_proxy,
services: HashMap::new(),
watch_services_task: None,
_events_task: events_task,
},
);
Ok(())
}
pub async fn disconnect_peripheral(&self, id: String) -> Result<(), Error> {
let peer_id = PeerId::from_str(&id)?;
self.inner.write().clients.remove(&peer_id);
Ok(())
}
pub fn get_central_proxy(&self) -> Option<CentralProxy> {
self.inner.read().central.as_ref().map(|c| c.proxy.clone())
}
fn get_remote_service_proxy(&self) -> Option<RemoteServiceProxy> {
self.inner.read().active_remote_service.as_ref().map(|s| s.proxy.clone())
}
pub fn get_scan_responses(&self) -> Vec<BleScanResponse> {
const EMPTY_DEVICE: &str = "";
let mut devices = Vec::new();
for (peer_id, peer) in &self.inner.read().scan_results {
let id = format!("{}", peer_id);
let name = peer.name.clone().unwrap_or_else(|| EMPTY_DEVICE.to_string());
let connectable = peer.connectable;
devices.push(BleScanResponse::new(id, name, connectable));
}
devices
}
pub fn print(&self) {
let tag = "GattClientFacade::print";
let inner = self.inner.read();
info!(
tag = &with_line!(tag),
"Central: {:?}, Active Service: {:?}, Scan Results: {:?}, Clients: {:?}",
inner.central,
inner.active_remote_service,
inner.scan_results,
inner.clients,
);
}
pub fn cleanup(&self) {
let mut inner = self.inner.write();
inner.active_remote_service = None;
inner.central = None;
inner.scan_results.clear();
inner.clients.clear();
inner.scan_task = None;
}
}