diff --git a/rust/bt-bap/src/types.rs b/rust/bt-bap/src/types.rs
index 65bb251..f3946d6 100644
--- a/rust/bt-bap/src/types.rs
+++ b/rust/bt-bap/src/types.rs
@@ -10,7 +10,7 @@
 
 /// Broadcast_ID is a 3-byte data on the wire.
 /// Defined in BAP spec v1.0.1 section 3.7.2.1.
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+#[derive(Clone, Copy, Eq, Hash, PartialEq)]
 pub struct BroadcastId(u32);
 
 impl BroadcastId {
@@ -24,7 +24,13 @@
 
 impl std::fmt::Display for BroadcastId {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "{:#08x}", self.0)
+        write!(f, "{:08x}", self.0)
+    }
+}
+
+impl std::fmt::Debug for BroadcastId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("BroadcastId").field(&format_args!("0x{}", self)).finish()
     }
 }
 
diff --git a/rust/bt-bass/src/client.rs b/rust/bt-bass/src/client.rs
index 8ad2ef6..fc86482 100644
--- a/rust/bt-bass/src/client.rs
+++ b/rust/bt-bass/src/client.rs
@@ -296,8 +296,8 @@
     ///   in
     /// * `pa_interval` - updated PA interval value. If none, unknown value is
     ///   used
-    /// * `bis_sync` - desired BIG to BIS synchronization information. If empty,
-    ///   it's not updated
+    /// * `bis_map` - desired BIG to BIS synchronization update information. If
+    ///   a BIG does not exist as a key, sync for that BIG is not updated.
     /// * `metadata_map` - map of updated metadata for BIGs. If a mapping does
     ///   not exist for a BIG, that BIG's metadata is not updated
     pub async fn modify_broadcast_source(
@@ -305,7 +305,7 @@
         broadcast_id: BroadcastId,
         pa_sync: PaSync,
         pa_interval: Option<PeriodicAdvertisingInterval>,
-        bis_sync: Option<HashMap<SubgroupIndex, BisSync>>,
+        bis_map: HashMap<SubgroupIndex, BisSync>,
         metadata_map: Option<HashMap<SubgroupIndex, Vec<Metadata>>>,
     ) -> Result<(), Error> {
         let op = {
@@ -314,13 +314,12 @@
                 .ok_or(Error::UnknownBroadcastSource(broadcast_id))?;
 
             // Update BIS_Sync param for BIGs if applicable.
-            if let Some(sync_map) = bis_sync {
-                for (big_index, group) in state.subgroups.iter_mut().enumerate() {
-                    if let Some(bis_sync) = sync_map.get(&(big_index as u8)) {
-                        group.bis_sync = bis_sync.clone();
-                    }
+            for (big_index, group) in state.subgroups.iter_mut().enumerate() {
+                if let Some(bis_sync) = bis_map.get(&(big_index as u8)) {
+                    group.bis_sync = bis_sync.clone();
                 }
             }
+
             // Update metadata for BIGs if applicable.
             if let Some(mut m) = metadata_map {
                 for (big_index, group) in state.subgroups.iter_mut().enumerate() {
@@ -779,7 +778,7 @@
             BroadcastId::try_from(0x11).unwrap(),
             PaSync::DoNotSync,
             Some(PeriodicAdvertisingInterval(0xAAAA)),
-            None,
+            HashMap::new(),
             None,
         );
         pin_mut!(op_fut);
@@ -827,7 +826,7 @@
             BroadcastId::try_from(0x11).unwrap(),
             PaSync::DoNotSync,
             None,
-            Some(HashMap::from([(0, BisSync::sync(vec![1, 3, 5]).unwrap())])),
+            HashMap::from([(0, BisSync::sync(vec![1, 3, 5]).unwrap())]),
             Some(HashMap::from([
                 (0, vec![Metadata::BroadcastAudioImmediateRenderingFlag]),
                 (1, vec![Metadata::Language("eng".to_string())]),
@@ -835,7 +834,7 @@
             ])),
         );
         pin_mut!(op_fut);
-        let polled = op_fut.poll_unpin(&mut noop_cx);
+        let polled: Poll<Result<(), Error>> = op_fut.poll_unpin(&mut noop_cx);
         assert_matches!(polled, Poll::Ready(Ok(_)));
     }
 
@@ -849,7 +848,7 @@
             BroadcastId::try_from(0x11).unwrap(),
             PaSync::DoNotSync,
             None,
-            None,
+            HashMap::new(),
             None,
         );
         pin_mut!(op_fut);
diff --git a/rust/bt-bass/src/client/event.rs b/rust/bt-bass/src/client/event.rs
index bf4349e..f7ad10b 100644
--- a/rust/bt-bass/src/client/event.rs
+++ b/rust/bt-bass/src/client/event.rs
@@ -122,91 +122,75 @@
 
         loop {
             if let Some(item) = self.event_queue.pop_front() {
-                match item {
-                    Ok(event) => return Poll::Ready(Some(Ok(event))),
-                    Err(e) => {
-                        // If an error was received, we terminate the event stream, but send an
-                        // error to indicate why it was terminated.
-                        self.terminated = true;
-                        return Poll::Ready(Some(Err(e)));
+                // An error from a single stream will be reported, but the main EventStream
+                // will continue for other sources.
+                return Poll::Ready(Some(item));
+            }
+
+            let Some(item) = futures::ready!(self.notification_streams.poll_next_unpin(cx)) else {
+                // All notification streams have been closed. Terminate the EventStream.
+                self.terminated = true;
+                let err = Error::EventStream(Box::new(Error::Service(
+                    ServiceError::NotificationChannelClosed(format!(
+                        "All BASS GATT notification streams closed"
+                    )),
+                )));
+                return Poll::Ready(Some(Err(err)));
+            };
+
+            // One of the notification streams produced an error. Report it, but do not
+            // terminate. SelectAll will remove the faulty stream from its set.
+            let Ok(notification) = item else {
+                let err = Error::EventStream(Box::new(Error::Gatt(item.unwrap_err())));
+                return Poll::Ready(Some(Err(err)));
+            };
+
+            let char_handle = notification.handle;
+            let (Ok(new_state), _) = BroadcastReceiveState::decode(notification.value.as_slice())
+            else {
+                self.event_queue.push_back(Ok(Event::UnknownPacket));
+                continue;
+            };
+
+            let maybe_prev_state = {
+                let mut lock = self.broadcast_sources.lock();
+                lock.update_state(char_handle, new_state.clone())
+            };
+
+            // If the previous value was not empty, check if it was overwritten.
+            if let Some(ref prev_state) = maybe_prev_state {
+                if let BroadcastReceiveState::NonEmpty(prev_receive_state) = prev_state {
+                    if new_state.is_empty() || !new_state.has_same_broadcast_id(&prev_state) {
+                        self.event_queue.push_back(Ok(Event::RemovedBroadcastSource(
+                            prev_receive_state.broadcast_id,
+                        )));
                     }
                 }
             }
 
-            match self.notification_streams.poll_next_unpin(cx) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(None) => {
-                    let err = Error::EventStream(Box::new(Error::Service(
-                        ServiceError::NotificationChannelClosed(format!(
-                            "GATT notification stream for BRS characteristics closed"
-                        )),
+            // BRS characteristic value was updated with a new broadcast source
+            // information.
+            if let BroadcastReceiveState::NonEmpty(receive_state) = &new_state {
+                let is_new_source = match maybe_prev_state {
+                    Some(prev_state) => !new_state.has_same_broadcast_id(&prev_state),
+                    None => true,
+                };
+                if is_new_source {
+                    self.event_queue.push_back(Ok(Event::AddedBroadcastSource(
+                        receive_state.broadcast_id,
+                        receive_state.pa_sync_state,
+                        receive_state.big_encryption,
                     )));
-                    self.event_queue.push_back(Err(err));
-                }
-                Poll::Ready(Some(Err(error))) => {
-                    // Deem all errors as critical.
-                    let err = Error::EventStream(Box::new(Error::Gatt(error)));
-                    self.event_queue.push_back(Err(err));
-                }
-                Poll::Ready(Some(Ok(notification))) => {
-                    let char_handle = notification.handle;
-                    let (Ok(new_state), _) =
-                        BroadcastReceiveState::decode(notification.value.as_slice())
-                    else {
-                        self.event_queue.push_back(Ok(Event::UnknownPacket));
-                        continue;
-                    };
-
-                    let maybe_prev_state = {
-                        let mut lock = self.broadcast_sources.lock();
-                        lock.update_state(char_handle, new_state.clone())
-                    };
-
-                    let mut multi_events = VecDeque::new();
-
-                    // If the previous value was not empty, check if it was overwritten.
-                    if let Some(ref prev_state) = maybe_prev_state {
-                        if let BroadcastReceiveState::NonEmpty(prev_receive_state) = prev_state {
-                            if new_state.is_empty() || !new_state.has_same_broadcast_id(&prev_state)
-                            {
-                                multi_events.push_back(Ok(Event::RemovedBroadcastSource(
-                                    prev_receive_state.broadcast_id,
-                                )));
-                            }
-                        }
+                } else {
+                    let other_events = Event::from_broadcast_receive_state(&receive_state);
+                    for e in other_events.into_iter() {
+                        self.event_queue.push_back(Ok(e));
                     }
-
-                    // BRS characteristic value was updated with a new broadcast source
-                    // information.
-                    if let BroadcastReceiveState::NonEmpty(receive_state) = &new_state {
-                        let is_new_source = match maybe_prev_state {
-                            Some(prev_state) => !new_state.has_same_broadcast_id(&prev_state),
-                            None => true,
-                        };
-                        if is_new_source {
-                            multi_events.push_back(Ok(Event::AddedBroadcastSource(
-                                receive_state.broadcast_id,
-                                receive_state.pa_sync_state,
-                                receive_state.big_encryption,
-                            )));
-                        } else {
-                            let other_events = Event::from_broadcast_receive_state(&receive_state);
-                            for e in other_events.into_iter() {
-                                multi_events.push_back(Ok(e));
-                            }
-                        }
-                    }
-                    if multi_events.len() != 0 {
-                        self.event_queue.append(&mut multi_events);
-                        continue;
-                    }
-                    continue;
                 }
-            };
-
-            break;
+            }
+            // Continue to the top of the loop to start draining the event_queue.
+            continue;
         }
-        Poll::Pending
     }
 }
 
@@ -220,7 +204,7 @@
     use futures::channel::mpsc::unbounded;
 
     use bt_common::core::AddressType;
-    use bt_gatt::types::Handle;
+    use bt_gatt::types::{Error as BtGattError, GattError, Handle};
 
     #[test]
     fn poll_event_stream() {
@@ -371,4 +355,57 @@
         // Should be pending because no more events generated from notifications.
         assert!(event_streams.poll_next_unpin(&mut noop_cx).is_pending());
     }
+
+    #[test]
+    fn error_on_one_stream_does_not_terminate_event_stream() {
+        let mut streams = SelectAll::new();
+        let (sender1, receiver1) = unbounded();
+        let (sender2, receiver2) = unbounded();
+        streams.push(receiver1.boxed());
+        streams.push(receiver2.boxed());
+
+        let source_tracker = Arc::new(Mutex::new(KnownBroadcastSources::new(HashMap::from([
+            (Handle(0x1), BroadcastReceiveState::Empty),
+            (Handle(0x2), BroadcastReceiveState::Empty),
+        ]))));
+        let mut event_streams = EventStream::new(streams, source_tracker);
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+
+        // Send an error on one stream.
+        sender1.unbounded_send(Err(BtGattError::Gatt(GattError::InvalidPdu))).expect("should send");
+
+        // We should receive the error event.
+        let polled = event_streams.poll_next_unpin(&mut noop_cx);
+        assert_matches!(polled, Poll::Ready(Some(Err(Error::EventStream(_)))));
+
+        // The stream should NOT be terminated.
+        assert!(!event_streams.is_terminated());
+        assert!(event_streams.poll_next_unpin(&mut noop_cx).is_pending());
+
+        // Send a valid notification on the other stream.
+        #[rustfmt::skip]
+        sender2
+            .unbounded_send(Ok(CharacteristicNotification {
+                handle: Handle(0x2),
+                value: vec![
+                    0x02, AddressType::Public as u8,             // source id and address type
+                    0x03, 0x04, 0x05, 0x06, 0x07, 0x08,          // address
+                    0x01, 0x02, 0x03, 0x04,                      // ad set id and broadcast id
+                    PaSyncState::Synced as u8,
+                    EncryptionStatus::NotEncrypted.raw_value(),
+                    0x00,                                        // no subgroups
+                ],
+                maybe_truncated: false,
+            }))
+            .expect("should send");
+
+        // We should be able to receive the event from the second stream.
+        let polled = event_streams.poll_next_unpin(&mut noop_cx);
+        assert_matches!(polled, Poll::Ready(Some(Ok(event))) => {
+            assert_eq!(event, Event::AddedBroadcastSource(BroadcastId::try_from(0x040302).unwrap(), PaSyncState::Synced, EncryptionStatus::NotEncrypted));
+        });
+
+        // The stream should still not be terminated.
+        assert!(!event_streams.is_terminated());
+    }
 }
diff --git a/rust/bt-broadcast-assistant/src/assistant/event.rs b/rust/bt-broadcast-assistant/src/assistant/event.rs
index d9a535a..18a97ae 100644
--- a/rust/bt-broadcast-assistant/src/assistant/event.rs
+++ b/rust/bt-broadcast-assistant/src/assistant/event.rs
@@ -106,40 +106,37 @@
 
         // 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(_)) => {
+        loop {
+            let Some(Ok(scanned)) = futures::ready!(self.scan_result_stream.poll_next_unpin(cx))
+            else {
                 self.terminated = true;
                 self.broadcast_source_scan_started.store(false, Ordering::Relaxed);
-                Poll::Ready(Some(Err(Error::CentralScanTerminated)))
+                return Poll::Ready(Some(Err(Error::CentralScanTerminated)));
+            };
+
+            let found_source = match Self::try_into_broadcast_source(&scanned) {
+                Err(error) => {
+                    return Poll::Ready(Some(Ok(Event::CouldNotParseAdvertisingData {
+                        peer: scanned.id,
+                        error,
+                    })));
+                }
+                Ok(None) => continue,
+                Ok(Some(source)) => 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,
+                })));
             }
         }
     }
@@ -153,33 +150,34 @@
 
     use bt_common::core::{AddressType, AdvertisingSetId};
     use bt_gatt::central::{AdvertisingDatum, PeerName};
-    use bt_gatt::test_utils::{FakeTypes, ScannedResultStream};
+    use bt_gatt::test_utils::{FakeTypes, ScannedResultStream, ScannedResultStreamController};
     use bt_gatt::types::Error as BtGattError;
     use bt_gatt::types::GattError;
 
-    fn setup_stream() -> (EventStream<FakeTypes>, ScannedResultStream) {
+    fn setup_stream() -> (EventStream<FakeTypes>, ScannedResultStreamController) {
         let fake_scan_result_stream = ScannedResultStream::new();
+        let controller = fake_scan_result_stream.controller();
         let broadcast_sources = DiscoveredBroadcastSources::new();
         let broadcast_source_scan_started = Arc::new(AtomicBool::new(false));
 
         (
             EventStream::<FakeTypes>::new(
-                fake_scan_result_stream.clone(),
+                fake_scan_result_stream,
                 broadcast_sources,
                 broadcast_source_scan_started,
             ),
-            fake_scan_result_stream,
+            controller,
         )
     }
 
     #[test]
     fn poll_found_broadcast_source_events() {
-        let (mut stream, mut scan_result_stream) = setup_stream();
+        let (mut stream, scan_result_controller) = setup_stream();
 
         // Scanned a broadcast source and its broadcast id.
         let broadcast_source_pid = PeerId(1005);
 
-        scan_result_stream.set_scanned_result(Ok(ScanResult {
+        scan_result_controller.add_scanned_result(Ok(ScanResult {
             id: broadcast_source_pid,
             connectable: true,
             name: PeerName::Unknown,
@@ -225,7 +223,7 @@
                     * (big #2 / bis #2) */
         ];
 
-        scan_result_stream.set_scanned_result(Ok(ScanResult {
+        scan_result_controller.add_scanned_result(Ok(ScanResult {
             id: broadcast_source_pid,
             connectable: true,
             name: PeerName::Unknown,
@@ -251,7 +249,7 @@
         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 {
+        scan_result_controller.add_scanned_result(Ok(ScanResult {
             id: broadcast_source_pid,
             connectable: true,
             name: PeerName::Unknown,
@@ -270,10 +268,10 @@
 
     #[test]
     fn central_scan_stream_terminates() {
-        let (mut stream, mut scan_result_stream) = setup_stream();
+        let (mut stream, scan_result_controller) = setup_stream();
 
         // Mimick scan error.
-        scan_result_stream.set_scanned_result(Err(BtGattError::Gatt(GattError::InvalidPdu)));
+        scan_result_controller.add_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) {
@@ -285,4 +283,68 @@
         assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Ready(None));
         assert_matches!(stream.is_terminated(), true);
     }
+
+    #[test]
+    fn poll_processes_multiple_results_eagerly() {
+        let (mut stream, scan_result_controller) = setup_stream();
+        let mut noop_cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
+
+        let broadcast_source_pid = PeerId(1005);
+
+        #[rustfmt::skip]
+        let base_data = vec![
+            0x10, 0x20, 0x30, 0x01, // presentation delay, num of subgroups
+            0x01, 0x03, 0x00, 0x00, 0x00, 0x00, // num of bis, codec id
+            0x00, // codec specific config len
+            0x00, // metadata len,
+            0x01, 0x00, // bis index, codec specific config len
+        ];
+
+        // 1. An irrelevant peer that should be ignored.
+        scan_result_controller.add_scanned_result(Ok(ScanResult {
+            id: PeerId(1),
+            connectable: true,
+            name: PeerName::Unknown,
+            advertised: vec![],
+            advertising_sid: Some(0),
+            periodic_advertising_interval: None,
+        }));
+        // 2. A broadcast source, but incomplete data (only broadcast ID).
+        scan_result_controller.add_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],
+            )],
+            advertising_sid: Some(1),
+            periodic_advertising_interval: None,
+        }));
+        // 3. The same broadcast source, with BASE data, which completes it.
+        scan_result_controller.add_scanned_result(Ok(ScanResult {
+            id: broadcast_source_pid,
+            connectable: true,
+            name: PeerName::Unknown,
+            advertised: vec![AdvertisingDatum::ServiceData(
+                BASIC_AUDIO_ANNOUNCEMENT_SERVICE,
+                base_data.clone(),
+            )],
+            advertising_sid: Some(1),
+            periodic_advertising_interval: None,
+        }));
+
+        // The stream should eagerly process all three items and emit the
+        // FoundBroadcastSource event.
+        let poll_result = stream.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(Ok(event))) = poll_result else {
+            panic!("should have received event, but got {:?}", poll_result);
+        };
+        assert_matches!(event, Event::FoundBroadcastSource{peer, ..} => {
+            assert_eq!(peer, broadcast_source_pid);
+        });
+
+        // The underlying stream is now empty, so the next poll should be pending.
+        assert!(stream.poll_next_unpin(&mut noop_cx).is_pending());
+    }
 }
diff --git a/rust/bt-broadcast-assistant/src/assistant/peer.rs b/rust/bt-broadcast-assistant/src/assistant/peer.rs
index 30be3f1..a901b21 100644
--- a/rust/bt-broadcast-assistant/src/assistant/peer.rs
+++ b/rust/bt-broadcast-assistant/src/assistant/peer.rs
@@ -160,7 +160,7 @@
             .unwrap_or(None);
 
         self.bass
-            .modify_broadcast_source(broadcast_id, pa_sync, pa_interval, Some(bis_sync), None)
+            .modify_broadcast_source(broadcast_id, pa_sync, pa_interval, bis_sync, None)
             .await
             .map_err(Into::into)
     }
diff --git a/rust/bt-broadcast-assistant/src/debug.rs b/rust/bt-broadcast-assistant/src/debug.rs
index 1fe4f9e..d8d26d8 100644
--- a/rust/bt-broadcast-assistant/src/debug.rs
+++ b/rust/bt-broadcast-assistant/src/debug.rs
@@ -18,7 +18,7 @@
 use bt_common::generic_audio::metadata_ltv::Metadata;
 use bt_common::PeerId;
 use bt_gatt::pii::GetPeerAddr;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use futures::stream::FusedStream;
 use futures::Future;
@@ -103,7 +103,7 @@
             return;
         };
         if let Err(e) = f(peer).await {
-            eprintln!("failed to perform oepration: {e:?}");
+            eprintln!("failed to perform operation: {e:?}");
         }
     }
 }
@@ -164,18 +164,45 @@
             eprintln!("Failed to parse big index from '{}', ignoring.", parts[0]);
             continue;
         };
-        let Ok(bis_index) = parse_int::<u8>(parts[1]) else {
-            eprintln!("Failed to parse bis index from '{}', ignoring.", parts[1]);
-            continue;
-        };
-        let entry = map.entry(ith_big).or_insert(BisSync::no_sync());
-        if let Err(e) = entry.synchronize_to_index(bis_index) {
-            eprintln!("Invalid BIS index: {e:?}");
+        match parse_int::<u8>(parts[1]) {
+            Ok(bis_index) => {
+                let entry = map.entry(ith_big).or_insert(BisSync::no_sync());
+                if let Err(e) = entry.synchronize_to_index(bis_index) {
+                    eprintln!("Failed to set sync to BIS index: {e:?}");
+                }
+            }
+            Err(_) if parts[1] == "OFF" => {
+                map.insert(ith_big, BisSync::no_sync());
+            }
+            Err(e) => {
+                eprintln!("{e:?} - BIS index should be a number from 1-31, ignoring {}", parts[1]);
+            }
         }
     }
     map
 }
 
+/// Converts a passcode string into a 16-byte broadcast code.
+/// The string is UTF-8 encoded and then padded with zeros on the right to a
+/// total length of 16 bytes. This result is a little-endian byte array
+/// equivalent to a 128-bit value.
+fn passcode_to_broadcast_code(passcode: &str) -> Result<[u8; 16], String> {
+    if passcode.is_empty() {
+        return Err("invalid broadcast code: passcode cannot be empty".to_string());
+    }
+    let code = passcode.as_bytes();
+    if code.len() > 16 {
+        return Err(format!(
+            "invalid broadcast code: '{}'. should be at max length 16, but was {}",
+            passcode,
+            code.len()
+        ));
+    }
+    let mut broadcast_code = [0u8; 16];
+    broadcast_code[..code.len()].copy_from_slice(code);
+    Ok(broadcast_code)
+}
+
 impl<T: bt_gatt::GattTypes + 'static, R: GetPeerAddr> CommandRunner for AssistantDebug<T, R>
 where
     <T as bt_gatt::GattTypes>::NotificationStream: std::marker::Send,
@@ -187,13 +214,18 @@
         cmd: Self::Set,
         args: Vec<String>,
     ) -> impl futures::Future<Output = Result<(), impl std::error::Error>> {
+        let help_subcommands: HashSet<&str> = HashSet::from(["help", "-h", "--help"]);
         async move {
+            if args.len() >= 1 && help_subcommands.contains(args[0].as_str()) {
+                eprintln!("usage: {}", cmd.help_simple());
+                return Ok(());
+            }
             match cmd {
                 AssistantCmd::Info => {
                     let known = self.assistant.known_broadcast_sources();
-                    eprintln!("Known Broadcast Sources:");
+                    println!("Known Broadcast Sources:");
                     for (id, s) in known {
-                        eprintln!("PeerId ({id}), source: {s:?}");
+                        println!("PeerId ({id}), source: {s:?}");
                     }
                 }
                 AssistantCmd::Connect => {
@@ -240,19 +272,16 @@
                         return Ok(());
                     };
 
-                    let code = args[1].as_bytes();
-                    if code.len() > 16 {
-                        eprintln!(
-                            "invalid broadcast code: {}. should be at max length 16",
-                            args[1]
-                        );
-                        return Ok(());
-                    }
-                    let mut passcode_vec = vec![0; 16];
-                    passcode_vec[16 - code.len()..16].copy_from_slice(code);
+                    let broadcast_code = match passcode_to_broadcast_code(&args[1]) {
+                        Ok(code) => code,
+                        Err(e) => {
+                            eprintln!("{e:?}");
+                            return Ok(());
+                        }
+                    };
+
                     self.with_peer(|peer| async move {
-                        peer.send_broadcast_code(broadcast_id, passcode_vec.try_into().unwrap())
-                            .await
+                        peer.send_broadcast_code(broadcast_id, broadcast_code).await
                     })
                     .await;
                 }
@@ -383,7 +412,7 @@
                         advertising_sid,
                     ) {
                         Ok(source) => {
-                            eprintln!("broadcast source after additional info: {source:?}")
+                            println!("broadcast source after additional info: {source:?}")
                         }
                         Err(e) => {
                             eprintln!("failed to enter in broadcast source information: {e:?}")
@@ -401,7 +430,7 @@
                     }
 
                     let Ok(peer_id) = parse_peer_id(&args[0]) else {
-                        println!("invalid peer id: {}", args[0]);
+                        eprintln!("invalid peer id: {}", args[0]);
                         return Ok(());
                     };
 
@@ -432,7 +461,7 @@
                         .assistant
                         .force_discover_broadcast_source_metadata(peer_id, all_big_metadata)
                     {
-                        Ok(source) => eprintln!("broadcast source with metadata: {source:?}"),
+                        Ok(source) => println!("broadcast source with metadata: {source:?}"),
                         Err(e) => eprintln!("failed to enter in broadcast source metadata: {e:?}"),
                     }
                 }
@@ -465,7 +494,7 @@
                         .assistant
                         .force_discover_broadcast_source_metadata(peer_id, all_big_metadata)
                     {
-                        Ok(source) => eprintln!("broadcast source with metadata: {source:?}"),
+                        Ok(source) => println!("broadcast source with metadata: {source:?}"),
                         Err(e) => {
                             eprintln!("failed to enter in empty broadcast source metadata: {e:?}")
                         }
@@ -524,11 +553,23 @@
 
     #[test]
     fn test_parse_bis_sync() {
+        // Basic case with multiple BIGs and BIS indices.
         let bis_sync = parse_bis_sync("0-1,0-2,1-1");
         assert_eq!(bis_sync.len(), 2);
         assert_eq!(bis_sync.get(&0), Some(&BisSync::sync(vec![1, 2]).unwrap()));
         assert_eq!(bis_sync.get(&1), Some(&BisSync::sync(vec![1]).unwrap()));
 
+        // Case with "OFF" to disable sync for a BIG.
+        let bis_sync = parse_bis_sync("0-1,1-OFF,0-2");
+        assert_eq!(bis_sync.len(), 2);
+        assert_eq!(bis_sync.get(&0), Some(&BisSync::sync(vec![1, 2]).unwrap()));
+        assert_eq!(bis_sync.get(&1), Some(&BisSync::no_sync()));
+
+        // Case where sync is set and then turned off for the same BIG.
+        let bis_sync = parse_bis_sync("0-5,0-OFF");
+        assert_eq!(bis_sync.len(), 1);
+        assert_eq!(bis_sync.get(&0), Some(&BisSync::no_sync()));
+
         // Will ignore invalid values.
         let bis_sync = parse_bis_sync("0-1,0-2,1:1,1-1-1,");
         assert_eq!(bis_sync.len(), 1);
@@ -537,4 +578,34 @@
         let bis_sync = parse_bis_sync("hellothisistoallynotvalid");
         assert_eq!(bis_sync.len(), 0);
     }
+
+    #[test]
+    fn test_passcode_to_broadcast_code() {
+        // UTF-8 string that is less than 16 bytes.
+        // Source of truth test case from Bluetooth Spec.
+        let code = "Børne House";
+        let expected = [
+            0x42, 0xc3, 0xb8, 0x72, 0x6e, 0x65, 0x20, 0x48, 0x6f, 0x75, 0x73, 0x65, 0x00, 0x00,
+            0x00, 0x00,
+        ];
+        let actual = passcode_to_broadcast_code(code).expect("should succeed");
+        assert_eq!(actual, expected);
+        assert_eq!(u128::from_le_bytes(actual), 0x00000000_6573756F_4820656E_72B8C342);
+
+        // Valid ASCII passcode, exactly 16 bytes.
+        let code = "1234567890123456";
+        let expected = [
+            0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x30, 0x31, 0x32, 0x33, 0x34,
+            0x35, 0x36,
+        ];
+        assert_eq!(passcode_to_broadcast_code(code).unwrap(), expected);
+
+        // Invalid passcode, over 16 bytes.
+        let code = "12345678901234567";
+        assert!(passcode_to_broadcast_code(code).is_err());
+
+        // Empty passcode should be an error.
+        let code = "";
+        assert!(passcode_to_broadcast_code(code).is_err());
+    }
 }
diff --git a/rust/bt-common/src/lib.rs b/rust/bt-common/src/lib.rs
index 382567d..1cb2cf0 100644
--- a/rust/bt-common/src/lib.rs
+++ b/rust/bt-common/src/lib.rs
@@ -13,7 +13,7 @@
 /// same peer as long as the PeerId was retrieved after the `Central` was
 /// instantiated. PeerIds can be valid longer than that (often if the peer is
 /// bonded)
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
 pub struct PeerId(pub u64);
 
 impl rust_core::fmt::Display for PeerId {
@@ -21,7 +21,13 @@
         &self,
         f: &mut rust_core::fmt::Formatter<'_>,
     ) -> std::result::Result<(), std::fmt::Error> {
-        write!(f, "{:x}", self.0)
+        write!(f, "{:016x}", self.0)
+    }
+}
+
+impl std::fmt::Debug for PeerId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("PeerId").field(&format_args!("0x{}", self)).finish()
     }
 }
 
diff --git a/rust/bt-gatt/src/test_utils.rs b/rust/bt-gatt/src/test_utils.rs
index d528b47..3397799 100644
--- a/rust/bt-gatt/src/test_utils.rs
+++ b/rust/bt-gatt/src/test_utils.rs
@@ -5,16 +5,17 @@
 use bt_common::core::{Address, AddressType};
 use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
 use futures::future::{ready, Ready};
-use futures::Stream;
+use futures::stream::{FusedStream, Stream};
 use parking_lot::Mutex;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
-use std::task::Poll;
+use std::task::{Poll, Waker};
 
 use bt_common::{PeerId, Uuid};
 
 use crate::central::ScanResult;
 use crate::client::CharacteristicNotification;
+
 use crate::periodic_advertising::{PeriodicAdvertising, SyncReport};
 use crate::pii::GetPeerAddr;
 use crate::server::{self, LocalService, ReadResponder, ServiceDefinition, WriteResponder};
@@ -221,24 +222,47 @@
     }
 }
 
-#[derive(Clone)]
+#[derive(Default, Debug)]
+struct ScannedResultStreamInner {
+    results: VecDeque<Result<ScanResult>>,
+    waker: Option<Waker>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct ScannedResultStreamController(Arc<Mutex<ScannedResultStreamInner>>);
+
+impl ScannedResultStreamController {
+    /// Add a single scanned result item to output from the stream.
+    pub fn add_scanned_result(&self, item: Result<ScanResult>) {
+        let mut lock = self.0.lock();
+        lock.results.push_back(item);
+        if let Some(waker) = lock.waker.take() {
+            waker.wake();
+        }
+    }
+}
+
+#[derive(Debug, Default)]
 pub struct ScannedResultStream {
     inner: Arc<Mutex<ScannedResultStreamInner>>,
 }
 
-#[derive(Default)]
-pub struct ScannedResultStreamInner {
-    result: Option<Result<ScanResult>>,
-}
-
 impl ScannedResultStream {
+    /// Creates a new ScannedResultStream.
+    /// Client can get a ScannedResultStreamController using the `controller`
+    /// method.
     pub fn new() -> Self {
-        Self { inner: Arc::new(Mutex::new(ScannedResultStreamInner::default())) }
+        Self::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);
+    pub fn controller(&self) -> ScannedResultStreamController {
+        ScannedResultStreamController(self.inner.clone())
+    }
+}
+
+impl FusedStream for ScannedResultStream {
+    fn is_terminated(&self) -> bool {
+        self.inner.lock().results.is_empty()
     }
 }
 
@@ -247,14 +271,15 @@
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,
-        _cx: &mut std::task::Context<'_>,
+        cx: &mut std::task::Context<'_>,
     ) -> Poll<Option<Self::Item>> {
         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
+        match lock.results.pop_front() {
+            Some(result) => Poll::Ready(Some(result)),
+            None => {
+                lock.waker = Some(cx.waker().clone());
+                Poll::Pending
+            }
         }
     }
 }
@@ -345,7 +370,7 @@
 
 impl crate::Central<FakeTypes> for FakeCentral {
     fn scan(&self, _filters: &[crate::central::ScanFilter]) -> ScannedResultStream {
-        ScannedResultStream::new()
+        ScannedResultStream::default()
     }
 
     fn connect(&self, peer_id: PeerId) -> <FakeTypes as GattTypes>::ConnectFuture {
diff --git a/rust/bt-gatt/src/tests.rs b/rust/bt-gatt/src/tests.rs
index 6cab78b..d8188e4 100644
--- a/rust/bt-gatt/src/tests.rs
+++ b/rust/bt-gatt/src/tests.rs
@@ -246,7 +246,8 @@
         advertising_sid: Some(0),
         periodic_advertising_interval: None,
     };
-    let _ = scan_results.set_scanned_result(Ok(scanned_result));
+    let controller = scan_results.controller();
+    let _ = controller.add_scanned_result(Ok(scanned_result));
 
     let polled = scan_results.poll_next_unpin(&mut noop_cx);
     assert_matches!(polled, Poll::Ready(Some(Ok(_))));