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

fuchsia_bt_test_affordances/
lib.rs
