rust/bt-broadcast-assistant: Use BroadcastSourceKey instead of raw id

Use the strongly-typed BroadcastSourceKey to track periodic
advertising futures, sync report streams, and active synchronization
states in EventStream.

Fixed: 542348813
Test: cargo test
Change-Id: Ic415f3a3c0798760ff9c5f4d396bcd8702872fb5
Reviewed-on: https://bluetooth-review.googlesource.com/c/bluetooth/+/3420

rust/bt-broadcast-assistant/src/assistant/event.rs

diff --git a/rust/bt-broadcast-assistant/src/assistant/event.rs b/rust/bt-broadcast-assistant/src/assistant/event.rs
index 656ff93..2240a36 100644
--- a/rust/bt-broadcast-assistant/src/assistant/event.rs
+++ b/rust/bt-broadcast-assistant/src/assistant/event.rs
@@ -23,7 +23,7 @@
 use bt_gatt::GattTypes;
 
 use crate::assistant::{
-    DiscoveredBroadcastSources, Error, BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
+    BroadcastSourceKey, DiscoveredBroadcastSources, Error, BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
     BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE,
 };
 use crate::types::BroadcastSource;
@@ -47,10 +47,10 @@
 >;
 
 type PeriodicAdvertisingFuture<T> =
-    Pin<Box<dyn Future<Output = (PeerId, u8, PeriodicAdvertisingSyncResult<T>)>>>;
+    Pin<Box<dyn Future<Output = (BroadcastSourceKey, PeriodicAdvertisingSyncResult<T>)>>>;
 
 type PeriodicAdvertisingStream =
-    Pin<Box<dyn Stream<Item = (PeerId, u8, Result<SyncReport, bt_gatt::types::Error>)>>>;
+    Pin<Box<dyn Stream<Item = (BroadcastSourceKey, Result<SyncReport, bt_gatt::types::Error>)>>>;
 
 /// A stream of discovered broadcast sources.
 /// This stream polls the scan results from GATT client to discover
@@ -66,7 +66,7 @@
 
     establishing_periodic_advertising_syncs: FuturesUnordered<PeriodicAdvertisingFuture<T>>,
     active_periodic_advertising_sync_streams: SelectAll<PeriodicAdvertisingStream>,
-    active_syncs: HashMap<(PeerId, u8), Option<AbortHandle>>,
+    active_syncs: HashMap<BroadcastSourceKey, Option<AbortHandle>>,
 }
 
 impl<T: bt_gatt::GattTypes> Unpin for EventStream<T> {}
@@ -105,12 +105,12 @@
         }
 
         match self.establishing_periodic_advertising_syncs.poll_next_unpin(cx) {
-            Poll::Ready(Some((peer_id, sid, Ok(stream)))) => {
-                self.handle_established_sync(peer_id, sid, stream);
+            Poll::Ready(Some((key, Ok(stream)))) => {
+                self.handle_established_sync(key, stream);
                 Poll::Ready(())
             }
-            Poll::Ready(Some((peer_id, sid, Err(_)))) => {
-                self.active_syncs.remove(&(peer_id, sid));
+            Poll::Ready(Some((key, Err(_)))) => {
+                self.active_syncs.remove(&key);
                 Poll::Ready(())
             }
             Poll::Ready(None) | Poll::Pending => Poll::Pending,
@@ -132,14 +132,14 @@
         }
 
         match self.active_periodic_advertising_sync_streams.poll_next_unpin(cx) {
-            Poll::Ready(Some((peer_id, sid, Ok(report)))) => {
-                match self.handle_periodic_advertising_report(peer_id, sid, report) {
+            Poll::Ready(Some((key, Ok(report)))) => {
+                match self.handle_periodic_advertising_report(key, report) {
                     Some(event) => Poll::Ready(Some(event)),
                     None => Poll::Ready(None), // Progressed, but no event
                 }
             }
-            Poll::Ready(Some((peer_id, sid, Err(_)))) => {
-                self.active_syncs.remove(&(peer_id, sid));
+            Poll::Ready(Some((key, Err(_)))) => {
+                self.active_syncs.remove(&key);
                 Poll::Ready(None) // Progressed, but no event
             }
             Poll::Ready(None) | Poll::Pending => Poll::Pending,
@@ -178,21 +178,19 @@
 
     fn handle_established_sync(
         &mut self,
-        peer_id: PeerId,
-        sid: u8,
+        key: BroadcastSourceKey,
         stream: <<T as GattTypes>::PeriodicAdvertising as PeriodicAdvertising>::SyncStream,
     ) {
-        let mapped_stream = stream.map(move |report| (peer_id, sid, report));
+        let mapped_stream = stream.map(move |report| (key, report));
         let (abortable_stream, abort_handle) = abortable(mapped_stream);
 
         self.active_periodic_advertising_sync_streams.push(Box::pin(abortable_stream));
-        self.active_syncs.insert((peer_id, sid), Some(abort_handle));
+        self.active_syncs.insert(key, Some(abort_handle));
     }
 
     fn handle_periodic_advertising_report(
         &mut self,
-        peer_id: PeerId,
-        sid: u8,
+        key: BroadcastSourceKey,
         report: SyncReport,
     ) -> Option<Event> {
         let SyncReport::PeriodicAdvertisingReport(report) = report else {
@@ -203,24 +201,17 @@
             return None;
         };
 
-        let Ok(advertising_sid) = AdvertisingSetId::try_from(sid) else {
-            return None;
-        };
-
-        let (broadcast_source, changed) = self.broadcast_sources.merge_broadcast_source_data(
-            &(peer_id, advertising_sid),
-            &BroadcastSource::default().with_endpoint(base),
-        );
+        let (broadcast_source, changed) = self
+            .broadcast_sources
+            .merge_broadcast_source_data(&key, &BroadcastSource::default().with_endpoint(base));
 
         if broadcast_source.is_ready_to_add() && changed {
-            if let Some(Some(handle)) =
-                self.active_syncs.remove(&(peer_id, advertising_sid.value()))
-            {
+            if let Some(Some(handle)) = self.active_syncs.remove(&key) {
                 handle.abort();
             }
             return Some(Event::FoundBroadcastSource {
-                peer: peer_id,
-                advertising_sid,
+                peer: key.0,
+                advertising_sid: key.1,
                 source: broadcast_source,
             });
         }
@@ -243,13 +234,14 @@
             return None;
         };
 
+        let key = (scanned.id, sid);
         let (broadcast_source, changed) =
-            self.broadcast_sources.merge_broadcast_source_data(&(scanned.id, sid), &found_source);
+            self.broadcast_sources.merge_broadcast_source_data(&key, &found_source);
 
         if broadcast_source.is_ready_to_add() && changed {
             return Some(Event::FoundBroadcastSource {
-                peer: scanned.id,
-                advertising_sid: sid,
+                peer: key.0,
+                advertising_sid: key.1,
                 source: broadcast_source,
             });
         }
@@ -269,7 +261,6 @@
 
         // If we are already actively syncing (or establishing a sync) for this
         // peer/SID, don't start another one.
-        let key = (scanned.id, sid.value());
         if self.active_syncs.contains_key(&key) {
             return None;
         }
@@ -280,7 +271,7 @@
             sid.value(),
             SyncConfiguration { filter_duplicates: true },
         );
-        let mapped_fut = fut.map(move |res| (scanned.id, sid.value(), res));
+        let mapped_fut = fut.map(move |res| (key, res));
         self.establishing_periodic_advertising_syncs.push(Box::pin(mapped_fut));
 
         None