fuchsia_bt_test_affordances/
lib.rs

1// Copyright 2025 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::anyhow;
6use fidl::endpoints::ClientEnd;
7use fidl_fuchsia_bluetooth::{PeerId, Uuid};
8use fidl_fuchsia_bluetooth_gatt2::{Characteristic, ServiceHandle, ServiceInfo};
9use fidl_fuchsia_bluetooth_le::{AdvertisingParameters, ConnectionMarker};
10use fidl_fuchsia_bluetooth_sys::{HostInfo, Peer};
11use fuchsia_async::LocalExecutor;
12use fuchsia_sync::Mutex;
13use futures::StreamExt;
14use futures::channel::{mpsc, oneshot};
15use std::sync::Arc;
16use std::thread;
17
18mod gatt;
19mod le;
20mod proxies;
21mod sys;
22
23use proxies::Proxies;
24
25// TODO(https://fxbug.dev/414848887): Return fidl_fuchsia_bluetooth_affordances::Error instead of
26// anyhow::Error.
27enum Request {
28    GetHosts(oneshot::Sender<Result<Vec<HostInfo>, anyhow::Error>>),
29    GetKnownPeers(oneshot::Sender<Result<Vec<Peer>, anyhow::Error>>),
30    GetPeerId([u8; 6], oneshot::Sender<Result<Option<PeerId>, anyhow::Error>>),
31    SetDiscovery(bool, oneshot::Sender<Result<(), anyhow::Error>>),
32    SetDiscoverability(bool, oneshot::Sender<Result<(), anyhow::Error>>),
33    SetConnectability(bool, oneshot::Sender<Result<(), anyhow::Error>>),
34    StartLeScan(
35        futures::channel::mpsc::UnboundedSender<
36            Vec<fidl_fuchsia_bluetooth_affordances::ScannedPeer>,
37        >,
38        oneshot::Sender<Result<(), anyhow::Error>>,
39    ),
40    StopLeScan(oneshot::Sender<Result<(), anyhow::Error>>),
41    ConnectLe(PeerId, oneshot::Sender<Result<(), anyhow::Error>>),
42    AdvertisePeripheral(
43        Box<AdvertisingParameters>,
44        std::time::Duration,
45        oneshot::Sender<Result<Option<PeerId>, anyhow::Error>>,
46    ),
47    PublishService(
48        Uuid,
49        ServiceHandle,
50        Vec<Characteristic>,
51        oneshot::Sender<Result<(), anyhow::Error>>,
52    ),
53    DiscoverServices(oneshot::Sender<Result<Vec<ServiceInfo>, anyhow::Error>>),
54    ReadCharacteristic(
55        ServiceHandle,
56        fidl_fuchsia_bluetooth_gatt2::Handle,
57        oneshot::Sender<Result<fidl_fuchsia_bluetooth_gatt2::ReadValue, anyhow::Error>>,
58    ),
59    RegisterCharacteristicNotifier(
60        ServiceHandle,
61        fidl_fuchsia_bluetooth_gatt2::Handle,
62        oneshot::Sender<Result<(), anyhow::Error>>,
63    ),
64    Stop,
65}
66
67pub struct WorkThread {
68    thread_handle: Mutex<Option<thread::JoinHandle<Result<(), anyhow::Error>>>>,
69    sender: mpsc::UnboundedSender<Request>,
70}
71
72impl WorkThread {
73    pub fn spawn() -> Self {
74        let (sender, receiver) = mpsc::unbounded::<Request>();
75
76        let thread_handle = thread::spawn(move || {
77            LocalExecutor::default().run_singlethreaded(Self::handle_requests(receiver))?;
78            Ok(())
79        });
80
81        Self { thread_handle: Mutex::new(Some(thread_handle)), sender }
82    }
83
84    async fn handle_requests(
85        mut receiver: mpsc::UnboundedReceiver<Request>,
86    ) -> Result<(), anyhow::Error> {
87        let mut proxies = Proxies::connect()?;
88        let mut host_cache: Vec<HostInfo> = Vec::new();
89        // TODO(https://fxbug.dev/396500079): Consider HashMap<PeerId, Peer> instead.
90        let peer_cache: Arc<Mutex<Vec<Peer>>> = Arc::new(Mutex::new(Vec::new()));
91        let mut _peripheral_connection: ClientEnd<ConnectionMarker>;
92
93        while let Some(request) = receiver.next().await {
94            match request {
95                Request::GetHosts(result_sender) => {
96                    if let Err(err) = sys::refresh_host_cache(&mut proxies, &mut host_cache).await {
97                        result_sender
98                            .send(Err(anyhow!("refresh_host_cache() error: {err}")))
99                            .unwrap();
100                        continue;
101                    }
102                    result_sender.send(Ok(host_cache.clone())).unwrap();
103                }
104                Request::GetKnownPeers(result_sender) => {
105                    if let Err(err) = sys::refresh_peer_cache(
106                        &mut proxies,
107                        std::time::Duration::from_millis(10),
108                        peer_cache.clone(),
109                    )
110                    .await
111                    {
112                        result_sender
113                            .send(Err(anyhow!("refresh_peer_cache() error: {err}")))
114                            .unwrap();
115                        continue;
116                    }
117                    result_sender.send(Ok(peer_cache.lock().clone())).unwrap();
118                }
119                Request::GetPeerId(address, result_sender) => {
120                    let result = sys::get_peer(
121                        &mut proxies,
122                        address,
123                        std::time::Duration::from_secs(2),
124                        peer_cache.clone(),
125                    )
126                    .await
127                    .map(|opt_peer| opt_peer.map(|peer| peer.id.unwrap()));
128                    result_sender.send(result).unwrap();
129                }
130                Request::SetDiscovery(discovery, result_sender) => {
131                    result_sender.send(sys::set_discovery(&mut proxies, discovery).await).unwrap();
132                }
133                Request::SetDiscoverability(discoverable, result_sender) => {
134                    result_sender
135                        .send(sys::set_discoverability(&mut proxies, discoverable).await)
136                        .unwrap();
137                }
138                Request::SetConnectability(connectable, result_sender) => {
139                    result_sender
140                        .send(sys::set_connectability(&proxies, connectable).await)
141                        .unwrap();
142                }
143                Request::StartLeScan(sender, result_sender) => {
144                    result_sender.send(le::start_le_scan(&mut proxies, sender).await).unwrap();
145                }
146                Request::StopLeScan(result_sender) => {
147                    let stopped = le::stop_scan(&proxies);
148                    if stopped {
149                        result_sender.send(Ok(())).unwrap();
150                    } else {
151                        result_sender.send(Err(anyhow!("No scan ongoing"))).unwrap();
152                    }
153                }
154                Request::ConnectLe(peer_id, result_sender) => {
155                    result_sender.send(le::connect_le(&mut proxies, &peer_id).await).unwrap();
156                }
157                Request::AdvertisePeripheral(parameters, timeout, result_sender) => {
158                    match le::advertise_peripheral(&proxies, *parameters, timeout).await {
159                        Ok(Some((peer_id, connection))) => {
160                            _peripheral_connection = connection;
161                            result_sender.send(Ok(Some(peer_id))).unwrap();
162                        }
163                        result => {
164                            result_sender.send(result.map(|_| None)).unwrap();
165                        }
166                    }
167                }
168                Request::PublishService(uuid, service_handle, characteristics, result_sender) => {
169                    match gatt::publish_service(&proxies, uuid, service_handle, characteristics)
170                        .await
171                    {
172                        Ok(mut local_service_request_stream) => {
173                            fuchsia_async::Task::spawn(async move {
174                                while let Some(Ok(request)) =
175                                    local_service_request_stream.next().await
176                                {
177                                    // Just log the request for now.
178                                    println!("Received LocalService request: {:?}", request);
179                                }
180                            })
181                            .detach();
182                            result_sender.send(Ok(())).unwrap();
183                        }
184                        Err(err) => {
185                            result_sender.send(Err(err)).unwrap();
186                        }
187                    }
188                }
189                Request::DiscoverServices(result_sender) => {
190                    result_sender.send(gatt::discover_services(&mut proxies).await).unwrap();
191                }
192                Request::ReadCharacteristic(
193                    service_handle,
194                    characteristic_handle,
195                    result_sender,
196                ) => {
197                    result_sender
198                        .send(
199                            gatt::read_characteristic(
200                                &proxies,
201                                service_handle,
202                                characteristic_handle,
203                            )
204                            .await,
205                        )
206                        .unwrap();
207                }
208                Request::RegisterCharacteristicNotifier(
209                    service_handle,
210                    characteristic_handle,
211                    result_sender,
212                ) => {
213                    result_sender
214                        .send(
215                            gatt::register_characteristic_notifier(
216                                &proxies,
217                                service_handle,
218                                characteristic_handle,
219                            )
220                            .await,
221                        )
222                        .unwrap();
223                }
224                Request::Stop => break,
225            }
226        }
227
228        Ok(())
229    }
230
231    pub fn join(&self) -> Result<(), anyhow::Error> {
232        self.sender.clone().unbounded_send(Request::Stop).unwrap();
233        if let Err(err) =
234            self.thread_handle.lock().take().unwrap().join().expect("Failed to join work thread")
235        {
236            return Err(anyhow!("Work thread exited with error: {err}"));
237        }
238        Ok(())
239    }
240
241    // Get hosts.
242    pub async fn get_hosts(&self) -> Result<Vec<HostInfo>, anyhow::Error> {
243        let (sender, receiver) = oneshot::channel::<Result<Vec<HostInfo>, anyhow::Error>>();
244        self.sender.clone().unbounded_send(Request::GetHosts(sender))?;
245        receiver.await?
246    }
247
248    // Get identifier of peer at `address`.
249    pub async fn get_peer_id(&self, address: [u8; 6]) -> Result<Option<PeerId>, anyhow::Error> {
250        let (sender, receiver) = oneshot::channel::<Result<Option<PeerId>, anyhow::Error>>();
251        self.sender.clone().unbounded_send(Request::GetPeerId(address, sender))?;
252        receiver.await?
253    }
254
255    pub async fn get_known_peers(&self) -> Result<Vec<Peer>, anyhow::Error> {
256        let (sender, receiver) = oneshot::channel::<Result<Vec<Peer>, anyhow::Error>>();
257        self.sender.clone().unbounded_send(Request::GetKnownPeers(sender))?;
258        receiver.await?
259    }
260
261    // Set discovery state.
262    pub async fn set_discovery(&self, discovery: bool) -> Result<(), anyhow::Error> {
263        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
264        self.sender.clone().unbounded_send(Request::SetDiscovery(discovery, sender))?;
265        receiver.await?
266    }
267
268    // Set discoverability state.
269    pub async fn set_discoverability(&self, discoverable: bool) -> Result<(), anyhow::Error> {
270        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
271        self.sender.clone().unbounded_send(Request::SetDiscoverability(discoverable, sender))?;
272        receiver.await?
273    }
274
275    // Set connection policy.
276    pub async fn set_connectability(&self, connectable: bool) -> Result<(), anyhow::Error> {
277        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
278        self.sender.clone().unbounded_send(Request::SetConnectability(connectable, sender))?;
279        receiver.await?
280    }
281
282    // Scan for nearby LE peripherals and broadcasters.
283    pub async fn start_le_scan(
284        &self,
285        sender: futures::channel::mpsc::UnboundedSender<
286            Vec<fidl_fuchsia_bluetooth_affordances::ScannedPeer>,
287        >,
288    ) -> Result<(), anyhow::Error> {
289        let (oneshot_sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
290        self.sender.clone().unbounded_send(Request::StartLeScan(sender, oneshot_sender))?;
291        receiver.await?
292    }
293
294    // Stop an ongoing LE scan. Returns an error if no scan is ongoing.
295    pub async fn stop_le_scan(&self) -> Result<(), anyhow::Error> {
296        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
297        self.sender.clone().unbounded_send(Request::StopLeScan(sender))?;
298        receiver.await?
299    }
300
301    // Connect an LE peer and store the connection.
302    pub async fn connect_le(&self, peer_id: PeerId) -> Result<(), anyhow::Error> {
303        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
304        self.sender.clone().unbounded_send(Request::ConnectLe(peer_id, sender))?;
305        receiver.await?
306    }
307
308    // Start advertising as an LE peripheral, accept the first connection, and return the PeerId of
309    // its initiator. If `connectable` is false, then advertise and return None.
310    pub async fn advertise_peripheral(
311        &self,
312        parameters: AdvertisingParameters,
313        timeout: std::time::Duration,
314    ) -> Result<Option<PeerId>, anyhow::Error> {
315        let (sender, receiver) = oneshot::channel::<Result<Option<PeerId>, anyhow::Error>>();
316        self.sender
317            .clone()
318            .unbounded_send(Request::AdvertisePeripheral(Box::new(parameters), timeout, sender))
319            .unwrap();
320        receiver.await?
321    }
322
323    // Publish a GATT service with the given parameters. GATT requests are logged.
324    pub async fn publish_service(
325        &self,
326        uuid: Uuid,
327        service_handle: ServiceHandle,
328        characteristics: Vec<Characteristic>,
329    ) -> Result<(), anyhow::Error> {
330        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
331        self.sender.clone().unbounded_send(Request::PublishService(
332            uuid,
333            service_handle,
334            characteristics,
335            sender,
336        ))?;
337        receiver.await?
338    }
339
340    // Discover the GATT services of the currently connected LE peer.
341    pub async fn discover_services(&self) -> Result<Vec<ServiceInfo>, anyhow::Error> {
342        let (sender, receiver) = oneshot::channel::<Result<Vec<ServiceInfo>, anyhow::Error>>();
343        self.sender.clone().unbounded_send(Request::DiscoverServices(sender))?;
344        receiver.await?
345    }
346
347    // Perform a short read of the GATT characteristic identified with the given handles.
348    pub async fn read_characteristic(
349        &self,
350        service_handle: ServiceHandle,
351        characteristic_handle: fidl_fuchsia_bluetooth_gatt2::Handle,
352    ) -> Result<fidl_fuchsia_bluetooth_gatt2::ReadValue, anyhow::Error> {
353        let (sender, receiver) =
354            oneshot::channel::<Result<fidl_fuchsia_bluetooth_gatt2::ReadValue, anyhow::Error>>();
355        self.sender.clone().unbounded_send(Request::ReadCharacteristic(
356            service_handle,
357            characteristic_handle,
358            sender,
359        ))?;
360        receiver.await?
361    }
362
363    // Enable notifications/indications on the GATT characteristic with the given handles.
364    //
365    // Only one operation on a Remote Service can be pending at a time.
366    pub async fn register_characteristic_notifier(
367        &self,
368        service_handle: ServiceHandle,
369        characteristic_handle: fidl_fuchsia_bluetooth_gatt2::Handle,
370    ) -> Result<(), anyhow::Error> {
371        let (sender, receiver) = oneshot::channel::<Result<(), anyhow::Error>>();
372        self.sender.clone().unbounded_send(Request::RegisterCharacteristicNotifier(
373            service_handle,
374            characteristic_handle,
375            sender,
376        ))?;
377        receiver.await?
378    }
379}
380
381#[cfg(test)]
382mod tests {
383    use super::*;
384
385    #[fuchsia::test]
386    fn test_update_peer_cache_handles_duplicates_in_input() {
387        let peer_cache = Arc::new(Mutex::new(Vec::new()));
388
389        let mut peer1 = Peer::default();
390        peer1.id = Some(PeerId { value: 1 });
391        peer1.name = Some("Peer 1".to_string());
392
393        let mut peer2 = Peer::default();
394        peer2.id = Some(PeerId { value: 1 });
395        peer2.name = Some("Peer 2".to_string());
396
397        // List of updated peers includes two entries with the same ID.
398        sys::update_peer_cache(peer_cache.clone(), vec![peer1, peer2.clone()], vec![]);
399
400        let cache = peer_cache.lock();
401
402        // The cache should only keep the final entry.
403        assert_eq!(cache.len(), 1);
404        assert_eq!(cache[0].name.as_deref(), Some("Peer 2"));
405    }
406
407    #[fuchsia::test]
408    fn test_update_peer_cache_replaces_existing_entry() {
409        let mut peer = Peer::default();
410        peer.id = Some(PeerId { value: 1 });
411        peer.name = Some("Peer".to_string());
412        let peer_cache = Arc::new(Mutex::new(vec![peer.clone()]));
413
414        // Update the peer currently inside the cache with a new name.
415        peer.name = Some("Updated peer".to_string());
416        sys::update_peer_cache(peer_cache.clone(), vec![peer], vec![]);
417
418        let cache = peer_cache.lock();
419
420        // The cache should only have one entry with the updated name.
421        assert_eq!(cache.len(), 1);
422        assert_eq!(cache[0].name.as_deref(), Some("Updated peer"));
423    }
424}
fuchsia_bt_test_affordances/lib.rs

fuchsia_bt_test_affordances/
lib.rs
