diff --git a/rust/Cargo.toml b/rust/Cargo.toml
index 60e0918..f4da3a7 100644
--- a/rust/Cargo.toml
+++ b/rust/Cargo.toml
@@ -12,6 +12,7 @@
 
 ## Local path dependencies (keep sorted)
 bt-bass = { path = "bt-bass" }
+bt-broadcast-assistant = { path = "bt-broadcast-assistant" }
 bt-common = { path = "bt-common" }
 bt-gatt = { path = "bt-gatt" }
 bt-pacs = { path = "bt-pacs" }
diff --git a/rust/bt-bass/src/client.rs b/rust/bt-bass/src/client.rs
index 3aa0b18..3479c66 100644
--- a/rust/bt-bass/src/client.rs
+++ b/rust/bt-bass/src/client.rs
@@ -13,7 +13,7 @@
 use parking_lot::Mutex;
 use tracing::warn;
 
-use bt_common::core::{AdvertisingSetId, PaInterval};
+use bt_common::core::{AddressType, AdvertisingSetId, PaInterval};
 use bt_common::generic_audio::metadata_ltv::Metadata;
 use bt_common::packet_encoding::Decodable;
 use bt_gatt::client::{CharacteristicNotification, PeerService, ServiceCharacteristic};
@@ -123,7 +123,7 @@
             audio_scan_control_point,
             broadcast_sources: Default::default(),
             broadcast_codes: HashMap::new(),
-            notification_streams: None,
+            notification_streams: Some(SelectAll::new()),
         }
     }
 
@@ -407,6 +407,11 @@
         }
         brs
     }
+
+    #[cfg(any(test, feature = "test-utils"))]
+    pub fn insert_broadcast_receive_state(&mut self, handle: Handle, brs: BroadcastReceiveState) {
+        self.broadcast_sources.lock().update_state(handle, brs);
+    }
 }
 
 #[cfg(test)]
diff --git a/rust/bt-bass/src/client/event.rs b/rust/bt-bass/src/client/event.rs
index 2c7ac2f..4077373 100644
--- a/rust/bt-bass/src/client/event.rs
+++ b/rust/bt-bass/src/client/event.rs
@@ -218,6 +218,7 @@
     use assert_matches::assert_matches;
     use futures::channel::mpsc::unbounded;
 
+    use bt_common::core::AddressType;
     use bt_gatt::types::Handle;
 
     #[test]
diff --git a/rust/bt-bass/src/types.rs b/rust/bt-bass/src/types.rs
index f810a3d..31c8b2d 100644
--- a/rust/bt-bass/src/types.rs
+++ b/rust/bt-bass/src/types.rs
@@ -3,7 +3,7 @@
 // found in the LICENSE file.
 
 use bt_common::core::ltv::LtValue;
-use bt_common::core::{AdvertisingSetId, PaInterval};
+use bt_common::core::{AddressType, AdvertisingSetId, PaInterval};
 use bt_common::generic_audio::metadata_ltv::*;
 use bt_common::packet_encoding::{Decodable, Encodable, Error as PacketError};
 use bt_common::{decodable_enum, Uuid};
@@ -676,32 +676,6 @@
     }
 }
 
-/// See Broadcast Audio Scan Service spec v1.0 Table 3.5 for details.
-#[repr(u8)]
-#[derive(Clone, Copy, Debug, PartialEq)]
-pub enum AddressType {
-    // Public Device Address or Public Identity Address.
-    Public = 0x00,
-    // Random Device Address or Random (static) Identity Address.
-    Random = 0x01,
-}
-
-impl AddressType {
-    const BYTE_SIZE: usize = 1;
-}
-
-impl TryFrom<u8> for AddressType {
-    type Error = PacketError;
-
-    fn try_from(value: u8) -> Result<Self, Self::Error> {
-        match value {
-            0x00 => Ok(Self::Public),
-            0x01 => Ok(Self::Random),
-            _ => Err(PacketError::OutOfRange),
-        }
-    }
-}
-
 /// Broadcast Receive State characteristic as defined in
 /// Broadcast Audio Scan Service spec v1.0 Section 3.2.
 /// The Broadcast Receive State characteristic is used by the server to expose
@@ -791,7 +765,7 @@
         + EncryptionStatus::MIN_PACKET_SIZE
         + NUM_SUBGROUPS_BYTE_SIZE;
 
-    #[cfg(test)]
+    #[cfg(any(test, feature = "test-utils"))]
     pub fn new(
         source_id: u8,
         source_address_type: AddressType,
diff --git a/rust/bt-broadcast-assistant/.gitignore b/rust/bt-broadcast-assistant/.gitignore
new file mode 100644
index 0000000..75e03aa
--- /dev/null
+++ b/rust/bt-broadcast-assistant/.gitignore
@@ -0,0 +1,17 @@
+# Generated by Cargo
+# will have compiled files and executables
+debug/
+target/
+
+# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
+# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
+Cargo.lock
+
+# These are backup files generated by rustfmt
+**/*.rs.bk
+
+# MSVC Windows builds of rustc generate these, which store debugging information
+*.pdb
+
+# Vim swap files
+*.swp
diff --git a/rust/bt-broadcast-assistant/Cargo.toml b/rust/bt-broadcast-assistant/Cargo.toml
new file mode 100644
index 0000000..a3ccfa6
--- /dev/null
+++ b/rust/bt-broadcast-assistant/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "bt-broadcast-assistant"
+version = "0.0.1"
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+bt-bass.workspace = true
+bt-common.workspace = true
+bt-gatt.workspace = true
+futures.workspace = true
+parking_lot.workspace = true
+thiserror.workspace = true
+
+[dev-dependencies]
+assert_matches.workspace = true
+bt-bass = { workspace = true, features = ["test-utils"] }
diff --git a/rust/bt-broadcast-assistant/src/assistant.rs b/rust/bt-broadcast-assistant/src/assistant.rs
new file mode 100644
index 0000000..3d3e61e
--- /dev/null
+++ b/rust/bt-broadcast-assistant/src/assistant.rs
@@ -0,0 +1,272 @@
+// Copyright 2023 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::Arc;
+use thiserror::Error;
+
+use bt_bass::client::error::Error as BassClientError;
+use bt_bass::client::BroadcastAudioScanServiceClient;
+use bt_bass::types::BroadcastId;
+use bt_common::{PeerId, Uuid};
+use bt_gatt::central::*;
+use bt_gatt::client::PeerServiceHandle;
+use bt_gatt::types::Error as GattError;
+use bt_gatt::Client;
+
+pub mod event;
+use event::*;
+pub mod peer;
+pub use peer::Peer;
+
+use crate::types::*;
+
+pub const BROADCAST_AUDIO_SCAN_SERVICE: Uuid = Uuid::from_u16(0x184F);
+pub const BASIC_AUDIO_ANNOUNCEMENT_SERVICE: Uuid = Uuid::from_u16(0x1851);
+pub const BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE: Uuid = Uuid::from_u16(0x1852);
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("GATT operation error: {0:?}")]
+    Gatt(#[from] GattError),
+
+    #[error("Broadcast Audio Scan Service client error at peer ({0}): {1:?}")]
+    BassClient(PeerId, BassClientError),
+
+    #[error("Not connected to Broadcast Audio Scan Service at peer ({0})")]
+    NotConnectedToBass(PeerId),
+
+    #[error("Central scanning terminated unexpectedly")]
+    CentralScanTerminated,
+
+    #[error("Failed to connect to service ({1}) at peer ({0})")]
+    ConnectionFailure(PeerId, Uuid),
+
+    #[error("Broadcast Assistant was already started previously. It cannot be started twice")]
+    AlreadyStarted,
+
+    #[error("Failed due to error: {0}")]
+    Generic(String),
+}
+
+/// Contains information about the currently-known broadcast
+/// sources and the peers they were found on
+#[derive(Debug)]
+pub(crate) struct DiscoveredBroadcastSources(Mutex<HashMap<PeerId, BroadcastSource>>);
+
+impl DiscoveredBroadcastSources {
+    /// Creates a shareable instance of `DiscoveredBroadcastSources`.
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self(Mutex::new(HashMap::new())))
+    }
+
+    /// Merges the broadcast source data with existing broadcast source data.
+    /// Returns the copy of the broadcast source data after the merge and
+    /// indicates whether it has changed from before or not.
+    pub(crate) fn merge_broadcast_source_data(
+        &self,
+        peer_id: &PeerId,
+        data: &BroadcastSource,
+    ) -> (BroadcastSource, bool) {
+        let mut lock = self.0.lock();
+        let source = lock.entry(*peer_id).or_default();
+        let before = source.clone();
+
+        source.merge(data);
+        let after = source.clone();
+        let changed = before != after;
+
+        (after, changed)
+    }
+
+    /// Get a BroadcastSource from a peer id.
+    fn get_by_peer_id(&self, peer_id: &PeerId) -> Option<BroadcastSource> {
+        let lock = self.0.lock();
+        lock.get(&peer_id).clone().map(|source| source.clone())
+    }
+
+    /// Get a BroadcastSource from associated broadcast id.
+    fn get_by_broadcast_id(&self, broadcast_id: &BroadcastId) -> Option<BroadcastSource> {
+        let lock = self.0.lock();
+        let info = lock.iter().find(|(&_k, &ref v)| v.broadcast_id == Some(*broadcast_id));
+        match info {
+            Some((&_peer_id, &ref broadcast_source)) => Some(broadcast_source.clone()),
+            None => None,
+        }
+    }
+}
+
+pub struct BroadcastAssistant<T: bt_gatt::GattTypes> {
+    central: T::Central,
+    broadcast_sources: Arc<DiscoveredBroadcastSources>,
+    scan_stream: Option<T::ScanResultStream>,
+}
+
+impl<T: bt_gatt::GattTypes + 'static> BroadcastAssistant<T> {
+    // Creates a broadcast assistant and sets it up to be ready
+    // for broadcast source scanning. Clients must use the `start`
+    // method to poll the event stream for scan results.
+    pub fn new(central: T::Central) -> Self
+    where
+        <T as bt_gatt::GattTypes>::ScanResultStream: std::marker::Send,
+    {
+        let scan_result_stream = central.scan(&Self::scan_filters());
+        Self {
+            central,
+            broadcast_sources: DiscoveredBroadcastSources::new(),
+            scan_stream: Some(scan_result_stream),
+        }
+    }
+
+    /// List of scan filters for advertisement data Broadcast Assistant should
+    /// look for, which are:
+    /// - Service data with Broadcast Audio Announcement Service UUID from
+    ///   Broadcast Sources (see BAP spec v1.0.1 Section 3.7.2.1 for details)
+    // TODO(b/308481381): define filter for finding broadcast sink.
+    fn scan_filters() -> Vec<ScanFilter> {
+        vec![Filter::HasServiceData(BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE).into()]
+    }
+
+    /// Start broadcast assistant. Returns EventStream that the upper layer can
+    /// poll. Upper layer can call methods on BroadcastAssistant based on the
+    /// events it sees.
+    pub fn start(&mut self) -> Result<EventStream, Error>
+    where
+        <T as bt_gatt::GattTypes>::ScanResultStream: std::marker::Send,
+    {
+        if self.scan_stream.is_none() {
+            return Err(Error::AlreadyStarted);
+        }
+        Ok(EventStream::new(self.scan_stream.take().unwrap(), self.broadcast_sources.clone()))
+    }
+
+    pub fn scan_for_scan_delegators(&mut self) -> T::ScanResultStream {
+        // Scan for service data with Broadcast Audio Scan Service UUID to look
+        // for Broadcast Sink collocated with the Scan Delegator (see BAP spec v1.0.1
+        // Section 3.9.2 for details).
+        self.central.scan(&vec![Filter::HasServiceData(BROADCAST_AUDIO_SCAN_SERVICE).into()])
+    }
+
+    pub async fn connect_to_scan_delegator(&mut self, peer_id: PeerId) -> Result<Peer<T>, Error>
+    where
+        <T as bt_gatt::GattTypes>::NotificationStream: std::marker::Send,
+    {
+        let client = self.central.connect(peer_id).await?;
+        let service_handles = client.find_service(BROADCAST_AUDIO_SCAN_SERVICE).await?;
+
+        for handle in service_handles {
+            if handle.uuid() != BROADCAST_AUDIO_SCAN_SERVICE || !handle.is_primary() {
+                continue;
+            }
+            let service = handle.connect().await?;
+            let bass = BroadcastAudioScanServiceClient::<T>::create(service)
+                .await
+                .map_err(|e| Error::BassClient(peer_id, e))?;
+
+            let connected_peer =
+                Peer::<T>::new(peer_id, client, bass, self.broadcast_sources.clone());
+            return Ok(connected_peer);
+        }
+        Err(Error::ConnectionFailure(peer_id, BROADCAST_AUDIO_SCAN_SERVICE))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use futures::{pin_mut, FutureExt};
+    use std::task::Poll;
+
+    use bt_common::core::{AddressType, AdvertisingSetId};
+    use bt_gatt::test_utils::{FakeCentral, FakeClient, FakeTypes};
+
+    use crate::assistant::peer::tests::fake_bass_service;
+
+    #[test]
+    fn merge_broadcast_source() {
+        let discovered = DiscoveredBroadcastSources::new();
+
+        let (bs, changed) = discovered.merge_broadcast_source_data(
+            &PeerId(1001),
+            &BroadcastSource::default()
+                .with_address([1, 2, 3, 4, 5, 6])
+                .with_address_type(AddressType::Public)
+                .with_advertising_sid(AdvertisingSetId(1))
+                .with_broadcast_id(BroadcastId(1001)),
+        );
+        assert!(changed);
+        assert_eq!(
+            bs,
+            BroadcastSource {
+                address: Some([1, 2, 3, 4, 5, 6]),
+                address_type: Some(AddressType::Public),
+                advertising_sid: Some(AdvertisingSetId(1)),
+                broadcast_id: Some(BroadcastId(1001)),
+                pa_interval: None,
+                endpoint: None,
+            }
+        );
+
+        let (bs, changed) = discovered.merge_broadcast_source_data(
+            &PeerId(1001),
+            &BroadcastSource::default().with_address_type(AddressType::Random).with_endpoint(
+                BroadcastAudioSourceEndpoint { presentation_delay_ms: 32, big: vec![] },
+            ),
+        );
+        assert!(changed);
+        assert_eq!(
+            bs,
+            BroadcastSource {
+                address: Some([1, 2, 3, 4, 5, 6]),
+                address_type: Some(AddressType::Random),
+                advertising_sid: Some(AdvertisingSetId(1)),
+                broadcast_id: Some(BroadcastId(1001)),
+                pa_interval: None,
+                endpoint: Some(BroadcastAudioSourceEndpoint {
+                    presentation_delay_ms: 32,
+                    big: vec![]
+                }),
+            }
+        );
+
+        let (_, changed) = discovered.merge_broadcast_source_data(
+            &PeerId(1001),
+            &BroadcastSource::default().with_address_type(AddressType::Random).with_endpoint(
+                BroadcastAudioSourceEndpoint { presentation_delay_ms: 32, big: vec![] },
+            ),
+        );
+        assert!(!changed);
+    }
+
+    #[test]
+    fn start_stream() {
+        let mut assistant = BroadcastAssistant::<FakeTypes>::new(FakeCentral::new());
+        let _ = assistant.start().expect("can start stream");
+
+        // Stream can only be started once.
+        assert!(assistant.start().is_err());
+    }
+
+    #[test]
+    fn connect_to_scan_delegator() {
+        // Set up fake GATT related objects.
+        let mut central = FakeCentral::new();
+        let mut client = FakeClient::new();
+        central.add_client(PeerId(1004), client.clone());
+        let service = fake_bass_service();
+        client.add_service(BROADCAST_AUDIO_SCAN_SERVICE, true, service.clone());
+
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+        let mut assistant = BroadcastAssistant::<FakeTypes>::new(central);
+        let conn_fut = assistant.connect_to_scan_delegator(PeerId(1004));
+        pin_mut!(conn_fut);
+        let polled = conn_fut.poll_unpin(&mut noop_cx);
+        let Poll::Ready(res) = polled else {
+            panic!("should be ready");
+        };
+        let _ = res.expect("should be ok");
+    }
+}
diff --git a/rust/bt-broadcast-assistant/src/assistant/event.rs b/rust/bt-broadcast-assistant/src/assistant/event.rs
new file mode 100644
index 0000000..993ba6f
--- /dev/null
+++ b/rust/bt-broadcast-assistant/src/assistant/event.rs
@@ -0,0 +1,259 @@
+// Copyright 2023 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use futures::stream::{FusedStream, Stream, StreamExt};
+use std::sync::Arc;
+use std::task::Poll;
+
+use bt_bass::types::BroadcastId;
+use bt_common::packet_encoding::Decodable;
+use bt_common::packet_encoding::Error as PacketError;
+use bt_common::PeerId;
+use bt_gatt::central::{AdvertisingDatum, ScanResult};
+use core::pin::Pin;
+
+use crate::assistant::{
+    DiscoveredBroadcastSources, Error, BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
+    BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE,
+};
+use crate::types::{BroadcastAudioSourceEndpoint, BroadcastSource};
+
+#[derive(Debug)]
+pub enum Event {
+    FoundBroadcastSource { peer: PeerId, source: BroadcastSource },
+    CouldNotParseAdvertisingData { peer: PeerId, error: PacketError },
+}
+
+/// A stream of discovered broadcast sources.
+/// This stream polls the scan results from GATT client to discover
+/// available broadcast sources.
+pub struct EventStream {
+    // Central scan result stream for finding broadcast sources.
+    scan_result_stream: Pin<Box<dyn Stream<Item = Result<ScanResult, bt_gatt::types::Error>>>>,
+    terminated: bool,
+
+    broadcast_sources: Arc<DiscoveredBroadcastSources>,
+}
+
+impl EventStream {
+    pub(crate) fn new(
+        scan_result_stream: impl Stream<Item = Result<ScanResult, bt_gatt::types::Error>> + 'static,
+        broadcast_sources: Arc<DiscoveredBroadcastSources>,
+    ) -> Self {
+        Self {
+            scan_result_stream: Box::pin(scan_result_stream),
+            terminated: false,
+            broadcast_sources,
+        }
+    }
+
+    /// Returns the broadcast source if the scanned peer is a broadcast source.
+    /// Returns an error if parsing of the scan result data fails and None if
+    /// the scanned peer is not a broadcast source.
+    fn try_into_broadcast_source(
+        scan_result: &ScanResult,
+    ) -> Result<Option<BroadcastSource>, PacketError> {
+        let mut source = None;
+        for datum in &scan_result.advertised {
+            let AdvertisingDatum::ServiceData(uuid, data) = datum else {
+                continue;
+            };
+            if *uuid == BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE {
+                let (bid, _) = BroadcastId::decode(data.as_slice())?;
+                source.get_or_insert(BroadcastSource::default()).with_broadcast_id(bid);
+            } else if *uuid == BASIC_AUDIO_ANNOUNCEMENT_SERVICE {
+                // TODO(dayeonglee): revisit when we implement periodic advertisement.
+                let (base, _) = BroadcastAudioSourceEndpoint::decode(data.as_slice())?;
+                source.get_or_insert(BroadcastSource::default()).with_endpoint(base);
+            }
+        }
+        Ok(source)
+    }
+}
+
+impl FusedStream for EventStream {
+    fn is_terminated(&self) -> bool {
+        self.terminated
+    }
+}
+
+impl Stream for EventStream {
+    type Item = Result<Event, Error>;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        if self.terminated {
+            return Poll::Ready(None);
+        }
+
+        // Poll scan result stream to check if there were any newly discovered peers
+        // that we're interested in.
+        match futures::ready!(self.scan_result_stream.poll_next_unpin(cx)) {
+            Some(Ok(scanned)) => {
+                match Self::try_into_broadcast_source(&scanned) {
+                    Err(e) => {
+                        return Poll::Ready(Some(Ok(Event::CouldNotParseAdvertisingData {
+                            peer: scanned.id,
+                            error: e,
+                        })));
+                    }
+                    Ok(Some(found_source)) => {
+                        // If we found a broadcast source, we add its information in the
+                        // internal records.
+                        let (broadcast_source, changed) = self
+                            .broadcast_sources
+                            .merge_broadcast_source_data(&scanned.id, &found_source);
+
+                        // Broadcast found event is relayed to the client iff complete
+                        // information has been gathered.
+                        if broadcast_source.into_add_source() && changed {
+                            return Poll::Ready(Some(Ok(Event::FoundBroadcastSource {
+                                peer: scanned.id,
+                                source: broadcast_source,
+                            })));
+                        }
+
+                        Poll::Pending
+                    }
+                    Ok(None) => Poll::Pending,
+                }
+            }
+            None | Some(Err(_)) => {
+                self.terminated = true;
+                Poll::Ready(Some(Err(Error::CentralScanTerminated)))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use assert_matches::assert_matches;
+
+    use bt_common::core::{AddressType, AdvertisingSetId};
+    use bt_gatt::central::{AdvertisingDatum, PeerName};
+    use bt_gatt::test_utils::ScannedResultStream;
+    use bt_gatt::types::Error as BtGattError;
+    use bt_gatt::types::GattError;
+
+    fn setup_stream() -> (EventStream, ScannedResultStream) {
+        let fake_scan_result_stream = ScannedResultStream::new();
+        let broadcast_sources = DiscoveredBroadcastSources::new();
+
+        (
+            EventStream::new(fake_scan_result_stream.clone(), broadcast_sources),
+            fake_scan_result_stream,
+        )
+    }
+
+    #[test]
+    fn poll_found_broadcast_source_events() {
+        let (mut stream, mut scan_result_stream) = setup_stream();
+
+        // Scanned a broadcast source and its broadcast id.
+        let broadcast_source_pid = PeerId(1005);
+
+        scan_result_stream.set_scanned_result(Ok(ScanResult {
+            id: broadcast_source_pid,
+            connectable: true,
+            name: PeerName::Unknown,
+            advertised: vec![AdvertisingDatum::ServiceData(
+                BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE,
+                vec![0x01, 0x02, 0x03],
+            )],
+        }));
+
+        // Found broadcast source event shouldn't have been sent since braodcast source
+        // information isn't complete.
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+        assert!(stream.poll_next_unpin(&mut noop_cx).is_pending());
+
+        // Pretend somehow address, address type, and advertising sid were
+        // filled out. This completes the broadcast source information.
+        // TODO(b/308481381): replace this block with sending a central scan result that
+        // contains the data.
+        let _ = stream.broadcast_sources.merge_broadcast_source_data(
+            &broadcast_source_pid,
+            &BroadcastSource::default()
+                .with_address([1, 2, 3, 4, 5, 6])
+                .with_address_type(AddressType::Public)
+                .with_advertising_sid(AdvertisingSetId(1)),
+        );
+
+        // Scanned broadcast source's BASE data.
+        // TODO(b/308481381): replace this block sending data through PA train instead.
+        #[rustfmt::skip]
+        let base_data = vec![
+            0x10, 0x20, 0x30, 0x02, // presentation delay, num of subgroups
+            0x01, 0x03, 0x00, 0x00, 0x00, 0x00, // num of bis, codec id (big #1)
+            0x00, // codec specific config len
+            0x00, // metadata len,
+            0x01, 0x00, // bis index, codec specific config len (big #1 / bis #1)
+            0x01, 0x02, 0x00, 0x00, 0x00, 0x00, // num of bis, codec id (big #2)
+            0x00, // codec specific config len
+            0x00, // metadata len,
+            0x01, 0x03, 0x02, 0x05,
+            0x08, /* bis index, codec specific config len, codec frame blocks LTV
+                    * (big #2 / bis #2) */
+        ];
+
+        scan_result_stream.set_scanned_result(Ok(ScanResult {
+            id: broadcast_source_pid,
+            connectable: true,
+            name: PeerName::Unknown,
+            advertised: vec![AdvertisingDatum::ServiceData(
+                BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
+                base_data.clone(),
+            )],
+        }));
+
+        // Expect the stream to send out broadcast source found event since information
+        // is complete.
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("should have received event");
+        };
+        assert_matches!(event, Event::FoundBroadcastSource{peer, ..} => {
+            assert_eq!(peer, broadcast_source_pid)
+        });
+
+        assert!(stream.poll_next_unpin(&mut noop_cx).is_pending());
+
+        // Scanned the same broadcast source's BASE data.
+        scan_result_stream.set_scanned_result(Ok(ScanResult {
+            id: broadcast_source_pid,
+            connectable: true,
+            name: PeerName::Unknown,
+            advertised: vec![AdvertisingDatum::ServiceData(
+                BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
+                base_data.clone(),
+            )],
+        }));
+
+        // Shouldn't have gotten the event again since the information remained the
+        // same.
+        assert!(stream.poll_next_unpin(&mut noop_cx).is_pending());
+    }
+
+    #[test]
+    fn central_scan_stream_terminates() {
+        let (mut stream, mut scan_result_stream) = setup_stream();
+
+        // Mimick scan error.
+        scan_result_stream.set_scanned_result(Err(BtGattError::Gatt(GattError::InvalidPdu)));
+
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+        match stream.poll_next_unpin(&mut noop_cx) {
+            Poll::Ready(Some(Err(e))) => assert_matches!(e, Error::CentralScanTerminated),
+            _ => panic!("should have received central scan terminated error"),
+        }
+
+        // Entire stream should have terminated.
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Ready(None));
+        assert_matches!(stream.is_terminated(), true);
+    }
+}
diff --git a/rust/bt-broadcast-assistant/src/assistant/peer.rs b/rust/bt-broadcast-assistant/src/assistant/peer.rs
new file mode 100644
index 0000000..3369a4d
--- /dev/null
+++ b/rust/bt-broadcast-assistant/src/assistant/peer.rs
@@ -0,0 +1,294 @@
+// Copyright 2023 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use futures::stream::FusedStream;
+use futures::Stream;
+use std::sync::Arc;
+use thiserror::Error;
+
+use bt_bass::client::error::Error as BassClientError;
+use bt_bass::client::event::Event as BassEvent;
+use bt_bass::client::{BigToBisSync, BroadcastAudioScanServiceClient};
+use bt_bass::types::{BroadcastId, PaSync};
+use bt_common::core::PaInterval;
+use bt_common::packet_encoding::Error as PacketError;
+use bt_common::PeerId;
+
+use crate::assistant::DiscoveredBroadcastSources;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("Failed to take event stream at Broadcast Audio Scan Service client")]
+    UnavailableBassEventStream,
+
+    #[error("Broadcast Audio Scan Service client error: {0:?}")]
+    BassClient(#[from] BassClientError),
+
+    #[error("Incomplete information for broadcast source with peer id ({0})")]
+    NotEnoughInfo(PeerId),
+
+    #[error("Broadcast source with peer id ({0}) does not exist")]
+    DoesNotExist(PeerId),
+
+    #[error("Packet error: {0}")]
+    PacketError(#[from] PacketError),
+}
+
+/// Connected scan delegator peer. Clients can use this
+/// object to perform Broadcast Audio Scan Service operations on the
+/// scan delegator peer.
+/// Not thread-safe and only one operation must be done at a time.
+pub struct Peer<T: bt_gatt::GattTypes> {
+    peer_id: PeerId,
+    // Keep for peer connection.
+    _client: T::Client,
+    bass: BroadcastAudioScanServiceClient<T>,
+    // TODO(b/309015071): add a field for pacs.
+    broadcast_sources: Arc<DiscoveredBroadcastSources>,
+}
+
+impl<T: bt_gatt::GattTypes> Peer<T> {
+    pub(crate) fn new(
+        peer_id: PeerId,
+        client: T::Client,
+        bass: BroadcastAudioScanServiceClient<T>,
+        broadcast_sources: Arc<DiscoveredBroadcastSources>,
+    ) -> Self {
+        Peer { peer_id, _client: client, bass, broadcast_sources }
+    }
+
+    pub fn peer_id(&self) -> PeerId {
+        self.peer_id
+    }
+
+    /// Takes event stream for BASS events from this scan delegator peer.
+    /// Clients can call this method to start subscribing to BASS events.
+    pub fn take_event_stream(
+        &mut self,
+    ) -> Result<impl Stream<Item = Result<BassEvent, BassClientError>> + FusedStream, Error> {
+        self.bass.take_event_stream().ok_or(Error::UnavailableBassEventStream)
+    }
+
+    /// Send broadcast code for a particular broadcast.
+    pub async fn send_broadcast_code(
+        &mut self,
+        broadcast_id: BroadcastId,
+        broadcast_code: [u8; 16],
+    ) -> Result<(), Error> {
+        self.bass.set_broadcast_code(broadcast_id, broadcast_code).await.map_err(Into::into)
+    }
+
+    /// Sends a command to add a particular broadcast source.
+    ///
+    /// # Arguments
+    ///
+    /// * `broadcast_source_pid` - peer id of the braodcast source that's to be
+    ///   added to this scan delegator peer
+    /// * `pa_sync` - pa sync mode the peer should attempt to be in
+    /// * `bis_sync` - desired BIG to BIS synchronization information. If the
+    ///   set is empty, no preference value is used for all the BIGs
+    pub async fn add_broadcast_source(
+        &mut self,
+        source_peer_id: PeerId,
+        pa_sync: PaSync,
+        bis_sync: BigToBisSync,
+    ) -> Result<(), Error> {
+        let broadcast_source = self
+            .broadcast_sources
+            .get_by_peer_id(&source_peer_id)
+            .ok_or(Error::DoesNotExist(source_peer_id))?;
+        if !broadcast_source.into_add_source() {
+            return Err(Error::NotEnoughInfo(source_peer_id));
+        }
+
+        self.bass
+            .add_broadcast_source(
+                broadcast_source.broadcast_id.unwrap(),
+                broadcast_source.address_type.unwrap(),
+                broadcast_source.address.unwrap(),
+                broadcast_source.advertising_sid.unwrap(),
+                pa_sync,
+                broadcast_source.pa_interval.unwrap_or(PaInterval::unknown()),
+                broadcast_source
+                    .endpoint
+                    .unwrap()
+                    .get_bass_subgroups(bis_sync)
+                    .map_err(Error::PacketError)?,
+            )
+            .await
+            .map_err(Into::into)
+    }
+
+    /// Sends a command to to update a particular broadcast source's PA sync.
+    ///
+    /// # Arguments
+    ///
+    /// * `broadcast_id` - broadcast id of the broadcast source that's to be
+    ///   updated
+    /// * `pa_sync` - pa sync mode the scan delegator peer should attempt to be
+    ///   in.
+    /// * `bis_sync` - desired BIG to BIS synchronization information
+    pub async fn update_broadcast_source_sync(
+        &mut self,
+        broadcast_id: BroadcastId,
+        pa_sync: PaSync,
+        bis_sync: BigToBisSync,
+    ) -> Result<(), Error> {
+        let pa_interval = self
+            .broadcast_sources
+            .get_by_broadcast_id(&broadcast_id)
+            .map(|bs| bs.pa_interval)
+            .unwrap_or(None);
+
+        self.bass
+            .modify_broadcast_source(broadcast_id, pa_sync, pa_interval, Some(bis_sync), None)
+            .await
+            .map_err(Into::into)
+    }
+
+    /// Sends a command to remove a particular broadcast source.
+    ///
+    /// # Arguments
+    ///
+    /// * `broadcast_id` - broadcast id of the braodcast source that's to be
+    ///   removed from the scan delegator
+    pub async fn remove_broadcast_source(
+        &mut self,
+        broadcast_id: BroadcastId,
+    ) -> Result<(), Error> {
+        self.bass.remove_broadcast_source(broadcast_id).await.map_err(Into::into)
+    }
+
+    /// Sends a command to inform the scan delegator peer that we have
+    /// started scanning for broadcast sources on behalf of it.
+    pub async fn inform_remote_scan_started(&mut self) -> Result<(), Error> {
+        self.bass.remote_scan_started().await.map_err(Into::into)
+    }
+
+    /// Sends a command to inform the scan delegator peer that we have
+    /// stopped scanning for broadcast sources on behalf of it.
+    pub async fn inform_remote_scan_stopped(&mut self) -> Result<(), Error> {
+        self.bass.remote_scan_stopped().await.map_err(Into::into)
+    }
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+
+    use assert_matches::assert_matches;
+    use futures::{pin_mut, FutureExt};
+    use std::collections::HashSet;
+    use std::task::Poll;
+
+    use bt_common::core::{AddressType, AdvertisingSetId};
+    use bt_gatt::test_utils::{FakeClient, FakePeerService, FakeTypes};
+    use bt_gatt::types::{
+        AttributePermissions, CharacteristicProperties, CharacteristicProperty, Handle,
+    };
+    use bt_gatt::Characteristic;
+
+    use crate::types::BroadcastSource;
+
+    const RECEIVE_STATE_HANDLE: Handle = Handle(0x11);
+    const AUDIO_SCAN_CONTROL_POINT_HANDLE: Handle = Handle(0x12);
+
+    pub(crate) fn fake_bass_service() -> FakePeerService {
+        let mut peer_service = FakePeerService::new();
+        // One broadcast receive state and one broadcast audio scan control
+        // point characteristic handles.
+        peer_service.add_characteristic(
+            Characteristic {
+                handle: RECEIVE_STATE_HANDLE,
+                uuid: bt_bass::types::BROADCAST_RECEIVE_STATE_UUID,
+                properties: CharacteristicProperties(vec![
+                    CharacteristicProperty::Broadcast,
+                    CharacteristicProperty::Notify,
+                ]),
+                permissions: AttributePermissions::default(),
+                descriptors: vec![],
+            },
+            vec![],
+        );
+        peer_service.add_characteristic(
+            Characteristic {
+                handle: AUDIO_SCAN_CONTROL_POINT_HANDLE,
+                uuid: bt_bass::types::BROADCAST_AUDIO_SCAN_CONTROL_POINT_UUID,
+                properties: CharacteristicProperties(vec![CharacteristicProperty::Broadcast]),
+                permissions: AttributePermissions::default(),
+                descriptors: vec![],
+            },
+            vec![],
+        );
+        peer_service
+    }
+
+    fn setup() -> (Peer<FakeTypes>, FakePeerService, Arc<DiscoveredBroadcastSources>) {
+        let peer_service = fake_bass_service();
+
+        let broadcast_sources = DiscoveredBroadcastSources::new();
+        (
+            Peer {
+                peer_id: PeerId(0x1),
+                _client: FakeClient::new(),
+                bass: BroadcastAudioScanServiceClient::<FakeTypes>::create_for_test(
+                    peer_service.clone(),
+                    Handle(0x1),
+                ),
+                broadcast_sources: broadcast_sources.clone(),
+            },
+            peer_service,
+            broadcast_sources,
+        )
+    }
+
+    #[test]
+    fn take_event_stream() {
+        let (mut peer, _peer_service, _broadcast_source) = setup();
+        let _event_stream = peer.take_event_stream().expect("should succeed");
+
+        // If we try to take the event stream the second time, it should fail.
+        assert!(peer.take_event_stream().is_err());
+    }
+
+    #[test]
+    fn add_broadcast_source_fail() {
+        let (mut peer, _peer_service, broadcast_source) = setup();
+
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+
+        // Should fail because broadcast source doesn't exist.
+        {
+            let fut = peer.add_broadcast_source(
+                PeerId(1001),
+                PaSync::SyncPastUnavailable,
+                HashSet::new(),
+            );
+            pin_mut!(fut);
+            let polled = fut.poll_unpin(&mut noop_cx);
+            assert_matches!(polled, Poll::Ready(Err(_)));
+        }
+
+        let _ = broadcast_source.merge_broadcast_source_data(
+            &PeerId(1001),
+            &BroadcastSource::default()
+                .with_address([1, 2, 3, 4, 5, 6])
+                .with_address_type(AddressType::Public)
+                .with_advertising_sid(AdvertisingSetId(1))
+                .with_broadcast_id(BroadcastId(1001)),
+        );
+
+        // Should fail because not enough information.
+        {
+            let fut = peer.add_broadcast_source(
+                PeerId(1001),
+                PaSync::SyncPastUnavailable,
+                HashSet::new(),
+            );
+            pin_mut!(fut);
+            let polled = fut.poll_unpin(&mut noop_cx);
+            assert_matches!(polled, Poll::Ready(Err(_)));
+        }
+    }
+}
diff --git a/rust/bt-broadcast-assistant/src/lib.rs b/rust/bt-broadcast-assistant/src/lib.rs
new file mode 100644
index 0000000..7d7af50
--- /dev/null
+++ b/rust/bt-broadcast-assistant/src/lib.rs
@@ -0,0 +1,7 @@
+// Copyright 2023 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+pub mod assistant;
+pub use assistant::BroadcastAssistant;
+pub mod types;
diff --git a/rust/bt-broadcast-assistant/src/types.rs b/rust/bt-broadcast-assistant/src/types.rs
new file mode 100644
index 0000000..8c6074e
--- /dev/null
+++ b/rust/bt-broadcast-assistant/src/types.rs
@@ -0,0 +1,435 @@
+// Copyright 2023 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use bt_bass::client::BigToBisSync;
+use bt_bass::types::{BigSubgroup, BisSync, BroadcastId};
+use bt_common::core::ltv::LtValue;
+use bt_common::core::{Address, AddressType, CodecId};
+use bt_common::core::{AdvertisingSetId, PaInterval};
+use bt_common::generic_audio::codec_configuration::CodecConfiguration;
+use bt_common::generic_audio::metadata_ltv::Metadata;
+use bt_common::packet_encoding::{Decodable, Error as PacketError};
+
+/// Broadcast source data as advertised through Basic Audio Announcement
+/// PA and Broadcast Audio Announcement.
+/// See BAP spec v1.0.1 Section 3.7.2.1 and Section 3.7.2.2 for details.
+// TODO(b/308481381): fill out endpoint from basic audio announcement from PA trains.
+#[derive(Clone, Default, Debug, PartialEq)]
+pub struct BroadcastSource {
+    pub(crate) address: Option<Address>,
+    pub(crate) address_type: Option<AddressType>,
+    pub(crate) advertising_sid: Option<AdvertisingSetId>,
+    pub(crate) broadcast_id: Option<BroadcastId>,
+    pub(crate) pa_interval: Option<PaInterval>,
+    pub(crate) endpoint: Option<BroadcastAudioSourceEndpoint>,
+}
+
+impl BroadcastSource {
+    /// Returns whether or not this BroadcastSource has enough information
+    /// to be added by the Broadcast Assistant.
+    pub(crate) fn into_add_source(&self) -> bool {
+        // PA interval is not necessary since default value can be used.
+        self.address.is_some()
+            && self.address_type.is_some()
+            && self.advertising_sid.is_some()
+            && self.broadcast_id.is_some()
+            && self.endpoint.is_some()
+    }
+
+    pub fn with_address(&mut self, address: [u8; 6]) -> &mut Self {
+        self.address = Some(address);
+        self
+    }
+
+    pub fn with_address_type(&mut self, type_: AddressType) -> &mut Self {
+        self.address_type = Some(type_);
+        self
+    }
+
+    pub fn with_broadcast_id(&mut self, bid: BroadcastId) -> &mut Self {
+        self.broadcast_id = Some(bid);
+        self
+    }
+
+    pub fn with_advertising_sid(&mut self, sid: AdvertisingSetId) -> &mut Self {
+        self.advertising_sid = Some(sid);
+        self
+    }
+
+    pub fn with_endpoint(&mut self, endpoint: BroadcastAudioSourceEndpoint) -> &mut Self {
+        self.endpoint = Some(endpoint);
+        self
+    }
+
+    /// Merge fields from other broadcast source into this broadcast source.
+    /// Set fields in other source take priority over this source.
+    /// If a field in the other broadcast source is none, it's ignored and
+    /// the existing values are kept.
+    pub(crate) fn merge(&mut self, other: &BroadcastSource) {
+        if let Some(address) = other.address {
+            self.address = Some(address);
+        }
+        if let Some(address_type) = other.address_type {
+            self.address_type = Some(address_type);
+        }
+        if let Some(advertising_sid) = other.advertising_sid {
+            self.advertising_sid = Some(advertising_sid);
+        }
+        if let Some(broadcast_id) = other.broadcast_id {
+            self.broadcast_id = Some(broadcast_id);
+        }
+        if let Some(pa_interval) = other.pa_interval {
+            self.pa_interval = Some(pa_interval);
+        }
+        if let Some(endpoint) = &other.endpoint {
+            self.endpoint = Some(endpoint.clone());
+        }
+    }
+}
+
+/// Parameters exposed as part of Basic Audio Announcement from Broadcast
+/// Sources. See BAP spec v1.0.1 Section 3.7.2.2 for more details.
+// TODO(b/308481381): Fill out the struct.
+#[derive(Clone, Debug, PartialEq)]
+pub struct BroadcastAudioSourceEndpoint {
+    // Delay is 3 bytes.
+    pub(crate) presentation_delay_ms: u32,
+    pub(crate) big: Vec<BroadcastIsochronousGroup>,
+}
+
+impl BroadcastAudioSourceEndpoint {
+    // Should contain presentation delay, num BIG, and at least one BIG praram.
+    const MIN_PACKET_SIZE: usize = 3 + 1 + BroadcastIsochronousGroup::MIN_PACKET_SIZE;
+
+    /// Returns the representation of this object's broadcast isochronous groups
+    /// that's usable with Broadcast Audio Scan Service operations.
+    ///
+    /// # Arguments
+    ///
+    /// * `bis_sync` - BIG to BIS sync information. If the set is empty, no
+    ///   preference value is used for all the BIGs
+    pub(crate) fn get_bass_subgroups(
+        &self,
+        bis_sync: BigToBisSync,
+    ) -> Result<Vec<BigSubgroup>, PacketError> {
+        let mut subgroups = Vec::new();
+        let sync_map = bt_bass::client::big_to_bis_sync_indices(&bis_sync);
+
+        for (big_index, group) in self.big.iter().enumerate() {
+            let bis_sync = match sync_map.get(&(big_index as u8)) {
+                Some(bis_indices) => {
+                    let mut bis_sync: BisSync = BisSync(0);
+                    bis_sync.set_sync(bis_indices)?;
+                    bis_sync
+                }
+                _ => BisSync::default(),
+            };
+            subgroups.push(BigSubgroup::new(Some(bis_sync)).with_metadata(group.metadata.clone()));
+        }
+        Ok(subgroups)
+    }
+}
+
+impl Decodable for BroadcastAudioSourceEndpoint {
+    type Error = PacketError;
+
+    fn decode(buf: &[u8]) -> core::result::Result<(Self, usize), Self::Error> {
+        if buf.len() < Self::MIN_PACKET_SIZE {
+            return Err(PacketError::UnexpectedDataLength);
+        }
+
+        let mut idx = 0 as usize;
+        let presentation_delay = u32::from_le_bytes([buf[idx], buf[idx + 1], buf[idx + 2], 0x00]);
+        idx += 3;
+
+        let num_big: usize = buf[idx] as usize;
+        idx += 1;
+        if num_big < 1 {
+            return Err(PacketError::InvalidParameter(format!(
+                "num of subgroups shall be at least 1 got {num_big}"
+            )));
+        }
+
+        let mut big = Vec::new();
+        while big.len() < num_big {
+            let (group, len) = BroadcastIsochronousGroup::decode(&buf[idx..])
+                .map_err(|e| PacketError::InvalidParameter(format!("{e}")))?;
+            big.push(group);
+            idx += len;
+        }
+
+        Ok((Self { presentation_delay_ms: presentation_delay, big }, idx))
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct BroadcastIsochronousGroup {
+    pub(crate) codec_id: CodecId,
+    pub(crate) codec_specific_config: Vec<CodecConfiguration>,
+    pub(crate) metadata: Vec<Metadata>,
+    pub(crate) bis: Vec<BroadcastIsochronousStream>,
+}
+
+impl BroadcastIsochronousGroup {
+    // Should contain num BIS, codec id, codec specific config len, metadata len,
+    // and at least one BIS praram.
+    const MIN_PACKET_SIZE: usize =
+        1 + CodecId::BYTE_SIZE + 1 + 1 + BroadcastIsochronousStream::MIN_PACKET_SIZE;
+}
+
+impl Decodable for BroadcastIsochronousGroup {
+    type Error = PacketError;
+
+    fn decode(buf: &[u8]) -> core::result::Result<(Self, usize), Self::Error> {
+        if buf.len() < BroadcastIsochronousGroup::MIN_PACKET_SIZE {
+            return Err(PacketError::UnexpectedDataLength);
+        }
+
+        let mut idx = 0;
+        let num_bis = buf[idx] as usize;
+        idx += 1;
+        if num_bis < 1 {
+            return Err(PacketError::InvalidParameter(format!(
+                "num of BIS shall be at least 1 got {num_bis}"
+            )));
+        }
+
+        let (codec_id, read_bytes) = CodecId::decode(&buf[idx..])?;
+        idx += read_bytes;
+
+        let codec_config_len = buf[idx] as usize;
+        idx += 1;
+
+        let (results, consumed) = CodecConfiguration::decode_all(&buf[idx..idx + codec_config_len]);
+        if consumed != codec_config_len {
+            return Err(bt_common::packet_encoding::Error::UnexpectedDataLength);
+        }
+
+        let codec_specific_configs = results.into_iter().filter_map(Result::ok).collect();
+        idx += codec_config_len;
+
+        let metadata_len = buf[idx] as usize;
+        idx += 1;
+
+        let (results_metadata, consumed_len) = Metadata::decode_all(&buf[idx..idx + metadata_len]);
+        if consumed_len != metadata_len {
+            return Err(PacketError::UnexpectedDataLength);
+        }
+        // Ignore any undecodable metadata types
+        let metadata = results_metadata.into_iter().filter_map(Result::ok).collect();
+        idx += consumed_len;
+
+        let mut bis = Vec::new();
+        while bis.len() < num_bis {
+            let (stream, len) = BroadcastIsochronousStream::decode(&buf[idx..])
+                .map_err(|e| PacketError::InvalidParameter(e.to_string()))?;
+            bis.push(stream);
+            idx += len;
+        }
+
+        Ok((
+            BroadcastIsochronousGroup {
+                codec_id,
+                codec_specific_config: codec_specific_configs,
+                metadata,
+                bis,
+            },
+            idx,
+        ))
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct BroadcastIsochronousStream {
+    pub(crate) bis_index: u8,
+    pub(crate) codec_specific_config: Vec<CodecConfiguration>,
+}
+
+impl BroadcastIsochronousStream {
+    const MIN_PACKET_SIZE: usize = 1 + 1;
+}
+
+impl Decodable for BroadcastIsochronousStream {
+    type Error = PacketError;
+
+    fn decode(buf: &[u8]) -> core::result::Result<(Self, usize), Self::Error> {
+        if buf.len() < BroadcastIsochronousStream::MIN_PACKET_SIZE {
+            return Err(PacketError::UnexpectedDataLength);
+        }
+
+        let mut idx = 0;
+
+        let bis_index = buf[idx];
+        idx += 1;
+
+        let codec_config_len = buf[idx] as usize;
+        idx += 1;
+
+        let (results, consumed) = CodecConfiguration::decode_all(&buf[idx..idx + codec_config_len]);
+        if consumed != codec_config_len {
+            return Err(bt_common::packet_encoding::Error::UnexpectedDataLength);
+        }
+        let codec_specific_configs = results.into_iter().filter_map(Result::ok).collect();
+        idx += codec_config_len;
+
+        Ok((
+            BroadcastIsochronousStream { bis_index, codec_specific_config: codec_specific_configs },
+            idx,
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashSet;
+
+    use super::*;
+
+    use bt_common::generic_audio::codec_configuration::{FrameDuration, SamplingFrequency};
+    use bt_common::generic_audio::AudioLocation;
+
+    #[test]
+    fn decode_bis() {
+        #[rustfmt::skip]
+        let buf = [
+            0x01, 0x09,                          // bis index and codec specific config len
+            0x02, 0x01, 0x06,                    // sampling frequency LTV
+            0x05, 0x03, 0x03, 0x00, 0x00, 0x0C,  // audio location LTV
+        ];
+
+        let (bis, _read_bytes) =
+            BroadcastIsochronousStream::decode(&buf[..]).expect("should not fail");
+        assert_eq!(
+            bis,
+            BroadcastIsochronousStream {
+                bis_index: 0x01,
+                codec_specific_config: vec![
+                    CodecConfiguration::SamplingFrequency(SamplingFrequency::F32000Hz),
+                    CodecConfiguration::AudioChannelAllocation(HashSet::from([
+                        AudioLocation::FrontLeft,
+                        AudioLocation::FrontRight,
+                        AudioLocation::LeftSurround,
+                        AudioLocation::RightSurround
+                    ])),
+                ],
+            }
+        );
+    }
+
+    #[test]
+    fn decode_big() {
+        #[rustfmt::skip]
+        let buf = [
+            0x02,  0x03, 0x00, 0x00, 0x00, 0x00,  // num of bis, codec id
+            0x04, 0x03, 0x04, 0x04, 0x10,         // codec specific config len, octets per codec frame LTV
+            0x03, 0x02, 0x08, 0x01,               // metadata len, audio active state LTV
+            0x01, 0x00,                           // bis index, codec specific config len (bis #1)
+            0x02, 0x03, 0x02, 0x02, 0x01,         // bis index, codec specific config len, frame duration LTV (bis #2)
+        ];
+
+        let (big, _read_bytes) =
+            BroadcastIsochronousGroup::decode(&buf[..]).expect("should not fail");
+        assert_eq!(
+            big,
+            BroadcastIsochronousGroup {
+                codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Transparent),
+                codec_specific_config: vec![CodecConfiguration::OctetsPerCodecFrame(0x1004),],
+                metadata: vec![Metadata::AudioActiveState(true)],
+                bis: vec![
+                    BroadcastIsochronousStream { bis_index: 0x01, codec_specific_config: vec![] },
+                    BroadcastIsochronousStream {
+                        bis_index: 0x02,
+                        codec_specific_config: vec![CodecConfiguration::FrameDuration(
+                            FrameDuration::TenMs
+                        )],
+                    },
+                ],
+            }
+        );
+    }
+
+    #[test]
+    fn decode_base() {
+        #[rustfmt::skip]
+        let buf = [
+            0x10, 0x20, 0x30, 0x02,               // presentation delay, num of subgroups
+            0x01, 0x03, 0x00, 0x00, 0x00, 0x00,   // num of bis, codec id (big #1)
+            0x00,                                 // codec specific config len
+            0x00,                                 // metadata len,
+            0x01, 0x00,                           // bis index, codec specific config len (big #1 / bis #1)
+            0x01, 0x02, 0x00, 0x00, 0x00, 0x00,   // num of bis, codec id (big #2)
+            0x00,                                 // codec specific config len
+            0x00,                                 // metadata len,
+            0x01, 0x03, 0x02, 0x05, 0x08,         // bis index, codec specific config len, codec frame blocks LTV (big #2 / bis #2)
+        ];
+
+        let (base, _read_bytes) =
+            BroadcastAudioSourceEndpoint::decode(&buf[..]).expect("should not fail");
+        assert_eq!(base.presentation_delay_ms, 0x00302010);
+        assert_eq!(base.big.len(), 2);
+        assert_eq!(
+            base.big[0],
+            BroadcastIsochronousGroup {
+                codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Transparent),
+                codec_specific_config: vec![],
+                metadata: vec![],
+                bis: vec![BroadcastIsochronousStream {
+                    bis_index: 0x01,
+                    codec_specific_config: vec![],
+                },],
+            }
+        );
+        assert_eq!(
+            base.big[1],
+            BroadcastIsochronousGroup {
+                codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Cvsd),
+                codec_specific_config: vec![],
+                metadata: vec![],
+                bis: vec![BroadcastIsochronousStream {
+                    bis_index: 0x01,
+                    codec_specific_config: vec![CodecConfiguration::CodecFramesPerSdu(0x08)],
+                },],
+            }
+        );
+    }
+
+    #[test]
+    fn decode_base_complex() {
+        // BroadcastAudioSourceEndpoint { presentation_delay_ms: 20000, big:
+        // [BroadcastIsochronousGroup { codec_id: Assigned(Lc3), codec_specific_config:
+        // [SamplingFrequency(F24000Hz), FrameDuration(TenMs),
+        // AudioChannelAllocation({FrontLeft, FrontRight}), OctetsPerCodecFrame(60)],
+        // metadata: [StreamingAudioContexts([Media])], bis:
+        // [BroadcastIsochronousStream { bis_index: 1, codec_specific_config:
+        // [AudioChannelAllocation({FrontLeft})] }, BroadcastIsochronousStream {
+        // bis_index: 2, codec_specific_con$
+        // ig: [AudioChannelAllocation({FrontRight})] }] }] }
+        #[rustfmt::skip]
+        let buf = [
+            0x20, 0x4e, 0x00, 0x01, 0x02, 0x06, 0x00, 0x00, 0x00, 0x00, 0x10, 0x02, 0x01, 0x05, 0x02, 0x02, 0x01, 0x05,
+            0x03, 0x03, 0x00, 0x00, 0x00, 0x03, 0x04, 0x3c, 0x00, 0x04, 0x03, 0x02, 0x04, 0x00, 0x01, 0x06, 0x05, 0x03, 0x01,
+            0x00, 0x00, 0x00, 0x02, 0x06, 0x05, 0x03, 0x02, 0x00, 0x00, 0x00
+        ];
+        let (base, _read_bytes) =
+            BroadcastAudioSourceEndpoint::decode(&buf[..]).expect("should not fail");
+        println!("{base:?}");
+
+        // BroadcastAudioSourceEndpoint { presentation_delay_ms: 20000, big:
+        // [BroadcastIsochronousGroup { codec_id: Assigned(Lc3), codec_specific_config:
+        // [SamplingFrequency(F24000Hz), FrameDuration(TenMs),
+        // AudioChannelAllocation({FrontLeft}), OctetsPerCodecFrame(60)],
+        // metadata: [BroadcastAudioImmediateRenderingFlag], bis:
+        // [BroadcastIsochronousStream { bis_index: 1, codec_specific_config:
+        // [SamplingFrequency(F24000Hz)] }] }] }
+        #[rustfmt::skip]
+        let buf = [
+            0x20, 0x4e, 0x00, 0x01, 0x01, 0x06, 0x00, 0x00, 0x00, 0x00, 0x10, 0x02, 0x01, 0x05,
+            0x02, 0x02, 0x01, 0x05, 0x03, 0x01, 0x00, 0x00, 0x00, 0x03, 0x04, 0x3c, 0x00, 0x02,
+            0x01, 0x09, 0x01, 0x03, 0x02, 0x01, 0x05,
+        ];
+        let (base, _read_bytes) =
+            BroadcastAudioSourceEndpoint::decode(&buf[..]).expect("should not fail");
+        println!("{base:?}");
+    }
+}
diff --git a/rust/bt-common/src/core.rs b/rust/bt-common/src/core.rs
index fd02299..9ee3fa0 100644
--- a/rust/bt-common/src/core.rs
+++ b/rust/bt-common/src/core.rs
@@ -7,8 +7,36 @@
 
 use crate::packet_encoding::{Encodable, Error as PacketError};
 
-/// Represents the Advertising Set ID which is 1 byte long.
-/// Follows little endian encoding.
+/// Bluetooth Device Address that uniquely identifies the device
+/// to another Bluetooth device.
+/// See Core spec v5.3 Vol 2, Part B section 1.2.
+pub type Address = [u8; 6];
+
+/// See Core spec v5.3 Vol 3, Part C section 15.1.1.
+#[repr(u8)]
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum AddressType {
+    Public = 0x00,
+    Random = 0x01,
+}
+
+impl AddressType {
+    pub const BYTE_SIZE: usize = 1;
+}
+
+impl TryFrom<u8> for AddressType {
+    type Error = PacketError;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0x00 => Ok(Self::Public),
+            0x01 => Ok(Self::Random),
+            _ => Err(PacketError::OutOfRange),
+        }
+    }
+}
+
+/// Advertising Set ID which is 1 byte long.
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct AdvertisingSetId(pub u8);
 
@@ -17,8 +45,7 @@
     pub const BYTE_SIZE: usize = 1;
 }
 
-/// Represents the SyncInfo Interval value which is 2 bytes long.
-/// Follows little endian encoding.
+/// SyncInfo Interval value which is 2 bytes long.
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct PaInterval(pub u16);
 
diff --git a/rust/bt-gatt/src/test_utils.rs b/rust/bt-gatt/src/test_utils.rs
index 879a0fe..7a3578f 100644
--- a/rust/bt-gatt/src/test_utils.rs
+++ b/rust/bt-gatt/src/test_utils.rs
@@ -12,7 +12,7 @@
 
 use bt_common::{PeerId, Uuid};
 
-use crate::central::{AdvertisingDatum, PeerName, ScanResult};
+use crate::central::ScanResult;
 use crate::client::CharacteristicNotification;
 use crate::{types::*, GattTypes};
 
@@ -142,47 +142,97 @@
     }
 }
 
-pub struct FakeServiceHandle {}
+#[derive(Clone)]
+pub struct FakeServiceHandle {
+    pub uuid: Uuid,
+    pub is_primary: bool,
+    pub fake_service: FakePeerService,
+}
 
 impl crate::client::PeerServiceHandle<FakeTypes> for FakeServiceHandle {
     fn uuid(&self) -> Uuid {
-        todo!()
+        self.uuid
     }
 
     fn is_primary(&self) -> bool {
-        todo!()
+        self.is_primary
     }
 
     fn connect(&self) -> <FakeTypes as GattTypes>::ServiceConnectFut {
-        futures::future::ready(Ok(FakePeerService::new()))
+        futures::future::ready(Ok(self.fake_service.clone()))
     }
 }
 
-pub struct FakeClient {}
+#[derive(Default)]
+struct FakeClientInner {
+    fake_services: Vec<FakeServiceHandle>,
+}
+
+#[derive(Clone)]
+pub struct FakeClient {
+    inner: Arc<Mutex<FakeClientInner>>,
+}
+
+impl FakeClient {
+    pub fn new() -> Self {
+        FakeClient { inner: Arc::new(Mutex::new(FakeClientInner::default())) }
+    }
+
+    /// Add a fake peer service to this client.
+    pub fn add_service(&mut self, uuid: Uuid, is_primary: bool, fake_service: FakePeerService) {
+        self.inner.lock().fake_services.push(FakeServiceHandle { uuid, is_primary, fake_service });
+    }
+}
 
 impl crate::Client<FakeTypes> for FakeClient {
     fn peer_id(&self) -> PeerId {
         todo!()
     }
 
-    fn find_service(&self, _uuid: Uuid) -> <FakeTypes as GattTypes>::FindServicesFut {
-        futures::future::ready(Ok::<Vec<FakeServiceHandle>, Error>(vec![FakeServiceHandle {}]))
+    fn find_service(&self, uuid: Uuid) -> <FakeTypes as GattTypes>::FindServicesFut {
+        let fake_services = &self.inner.lock().fake_services;
+        let mut filtered_services = Vec::new();
+        for handle in fake_services {
+            if handle.uuid == uuid {
+                filtered_services.push(handle.clone());
+            }
+        }
+
+        futures::future::ready(Ok(filtered_services))
     }
 }
 
-pub struct SingleResultStream {
-    result: Option<Result<crate::central::ScanResult>>,
+#[derive(Clone)]
+pub struct ScannedResultStream {
+    inner: Arc<Mutex<ScannedResultStreamInner>>,
 }
 
-impl Stream for SingleResultStream {
+#[derive(Default)]
+pub struct ScannedResultStreamInner {
+    result: Option<Result<ScanResult>>,
+}
+
+impl ScannedResultStream {
+    pub fn new() -> Self {
+        Self { inner: Arc::new(Mutex::new(ScannedResultStreamInner::default())) }
+    }
+
+    /// Set scanned result item to output from the stream.
+    pub fn set_scanned_result(&mut self, item: Result<ScanResult>) {
+        self.inner.lock().result = Some(item);
+    }
+}
+
+impl Stream for ScannedResultStream {
     type Item = Result<ScanResult>;
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,
         _cx: &mut std::task::Context<'_>,
     ) -> Poll<Option<Self::Item>> {
-        if self.result.is_some() {
-            Poll::Ready(self.get_mut().result.take())
+        let mut lock = self.inner.lock();
+        if lock.result.is_some() {
+            Poll::Ready(lock.result.take())
         } else {
             // Never wake up, as if we never find another result
             Poll::Pending
@@ -194,7 +244,7 @@
 
 impl GattTypes for FakeTypes {
     type Central = FakeCentral;
-    type ScanResultStream = SingleResultStream;
+    type ScanResultStream = ScannedResultStream;
     type Client = FakeClient;
     type ConnectFuture = Ready<Result<FakeClient>>;
     type PeerServiceHandle = FakeServiceHandle;
@@ -208,21 +258,36 @@
 }
 
 #[derive(Default)]
-pub struct FakeCentral {}
+pub struct FakeCentralInner {
+    clients: HashMap<PeerId, FakeClient>,
+}
 
-impl crate::Central<FakeTypes> for FakeCentral {
-    fn scan(&self, _filters: &[crate::central::ScanFilter]) -> SingleResultStream {
-        SingleResultStream {
-            result: Some(Ok(ScanResult {
-                id: PeerId(1),
-                connectable: true,
-                name: PeerName::CompleteName("Marie's Pixel 7 Pro".to_owned()),
-                advertised: vec![AdvertisingDatum::Services(vec![Uuid::from_u16(0x1844)])],
-            })),
-        }
+#[derive(Clone)]
+pub struct FakeCentral {
+    inner: Arc<Mutex<FakeCentralInner>>,
+}
+
+impl FakeCentral {
+    pub fn new() -> Self {
+        Self { inner: Arc::new(Mutex::new(FakeCentralInner::default())) }
     }
 
-    fn connect(&self, _peer_id: PeerId) -> <FakeTypes as GattTypes>::ConnectFuture {
-        futures::future::ready(Ok(FakeClient {}))
+    pub fn add_client(&mut self, peer_id: PeerId, client: FakeClient) {
+        let _ = self.inner.lock().clients.insert(peer_id, client);
+    }
+}
+
+impl crate::Central<FakeTypes> for FakeCentral {
+    fn scan(&self, _filters: &[crate::central::ScanFilter]) -> ScannedResultStream {
+        ScannedResultStream::new()
+    }
+
+    fn connect(&self, peer_id: PeerId) -> <FakeTypes as GattTypes>::ConnectFuture {
+        let clients = &self.inner.lock().clients;
+        let res = match clients.get(&peer_id) {
+            Some(client) => Ok(client.clone()),
+            None => Err(Error::PeerDisconnected(peer_id)),
+        };
+        futures::future::ready(res)
     }
 }
diff --git a/rust/bt-gatt/src/tests.rs b/rust/bt-gatt/src/tests.rs
index d300f0f..e753706 100644
--- a/rust/bt-gatt/src/tests.rs
+++ b/rust/bt-gatt/src/tests.rs
@@ -9,6 +9,7 @@
 
 use bt_common::{PeerId, Uuid};
 
+use crate::central::{AdvertisingDatum, PeerName, ScanResult};
 use crate::test_utils::*;
 use crate::types::*;
 use crate::{central::Filter, client::PeerService, Central};
@@ -18,7 +19,7 @@
 const TEST_UUID_3: Uuid = Uuid::from_u16(0x3456);
 
 // Sets up a fake peer service with some characteristics.
-fn set_up() -> FakePeerService {
+fn setup_peer_service() -> FakePeerService {
     let mut fake_peer_service = FakePeerService::new();
     fake_peer_service.add_characteristic(
         Characteristic {
@@ -86,7 +87,7 @@
 fn peer_service_discover_characteristics_works() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
 
-    let fake_peer_service = set_up();
+    let fake_peer_service = setup_peer_service();
 
     let mut discover_results = fake_peer_service.discover_characteristics(Some(TEST_UUID_1));
     let polled = discover_results.poll_unpin(&mut noop_cx);
@@ -110,7 +111,7 @@
 fn peer_service_read_characteristic() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
 
-    let mut fake_peer_service = set_up();
+    let mut fake_peer_service = setup_peer_service();
     let mut buf = vec![0; 255];
 
     // For characteristic that was added, value is returned
@@ -152,7 +153,7 @@
 fn peer_service_write_characteristic() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
 
-    let mut fake_peer_service: FakePeerService = set_up();
+    let mut fake_peer_service: FakePeerService = setup_peer_service();
 
     // Set expected characteristic value before calling `write_characteristic`.
     fake_peer_service.expect_characteristic_value(&Handle(1), vec![0, 1, 2, 3]);
@@ -174,7 +175,7 @@
 fn peer_service_write_characteristic_fail() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
 
-    let fake_peer_service = set_up();
+    let fake_peer_service = setup_peer_service();
 
     // Write some random value without calling `expect_characteristic_value` first.
     let mut write_result = fake_peer_service.write_characteristic(
@@ -194,7 +195,7 @@
 fn peer_service_subsribe() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
 
-    let mut fake_peer_service = set_up();
+    let mut fake_peer_service = setup_peer_service();
     let mut notification_stream = fake_peer_service.subscribe(&Handle(0x1));
 
     // Stream is empty unless we add an item through the FakeNotificationStream
@@ -230,9 +231,19 @@
 #[test]
 fn central_search_works() {
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
-    let central = FakeCentral::default();
+    let central = FakeCentral::new();
 
     let mut scan_results = central.scan(&[Filter::ServiceUuid(Uuid::from_u16(0x1844)).into()]);
+    let polled = scan_results.poll_next_unpin(&mut noop_cx);
+    assert_matches!(polled, Poll::Pending);
+
+    let scanned_result = ScanResult {
+        id: PeerId(1),
+        connectable: true,
+        name: PeerName::CompleteName("Marie's Pixel 7 Pro".to_owned()),
+        advertised: vec![AdvertisingDatum::Services(vec![Uuid::from_u16(0x1844)])],
+    };
+    let _ = scan_results.set_scanned_result(Ok(scanned_result));
 
     let polled = scan_results.poll_next_unpin(&mut noop_cx);
     assert_matches!(polled, Poll::Ready(Some(Ok(_))));
@@ -243,8 +254,8 @@
 
     let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
     let _stream = central.scan(&[]);
-    let connect_fut = central.connect(PeerId(1));
 
+    let connect_fut = central.connect(PeerId(1));
     futures::pin_mut!(connect_fut);
     let Poll::Ready(Ok(client)) = connect_fut.poll(&mut noop_cx) else {
         panic!("Connect should be ready Ok");
@@ -273,7 +284,12 @@
 
 #[test]
 fn central_dynamic_usage() {
-    let central: Box<dyn crate::Central<FakeTypes>> = Box::new(FakeCentral::default());
+    let mut central = FakeCentral::new();
+    let mut fake_client = FakeClient::new();
+    fake_client.add_service(Uuid::from_u16(0), true, FakePeerService::new());
+    central.add_client(PeerId(1), fake_client);
 
-    boxed_generic_usage(central);
+    let boxed: Box<dyn crate::Central<FakeTypes>> = Box::new(central);
+
+    boxed_generic_usage(boxed);
 }