rust/bt-bass: Handle state updates and send GATT notifications Add public APIs to add, update, and clear Broadcast Receive State characteristics. These will be used in response to any `ServerEvents` generated when the remote Broadcast Assistant requests an operation. Dispatch GATT notifications to subscribed peers when characteristic state is updated. Bug: 534436439 Test: cargo test, ./presubmit.sh Change-Id: I6a28b1077383b7d8ec5213f33b83067fe9c6437e Reviewed-on: https://bluetooth-review.googlesource.com/c/bluetooth/+/3220
diff --git a/rust/bt-bass/src/server.rs b/rust/bt-bass/src/server.rs index 715a803..0d9f529 100644 --- a/rust/bt-bass/src/server.rs +++ b/rust/bt-bass/src/server.rs
@@ -17,6 +17,7 @@ use bt_gatt::Characteristic; use futures::stream::Stream; use futures::Future; +use log::warn; use pin_project::pin_project; use std::collections::{BTreeMap, HashMap}; use std::num::NonZeroU8; @@ -40,13 +41,30 @@ /// Internal representation of a Broadcast Receive State characteristic slot. /// See BASS v1.0 Section 3.2. #[derive(Debug)] -pub(crate) struct PublishedReceiveStateCharacteristic { +struct PublishedReceiveStateCharacteristic { /// Handle assigned to this GATT characteristic. handle: Handle, /// Current Broadcast Receive State value. state: BroadcastReceiveState, } +/// A pending GATT notification to be dispatched following a state modification. +#[derive(Debug, PartialEq, Eq)] +struct PendingNotification { + /// Handle of the characteristic to notify. + handle: Handle, + /// Encoded GATT characteristic value. + value: Vec<u8>, +} + +/// A parsed request to write to the Control Point characteristic. +struct ControlPointWriteResult { + /// The decoded event associated with this request. + event: ServerEvent, + /// An optional notification to be sent to GATT peers. + notification: Option<PendingNotification>, +} + /// Internal publication state lifecycle of the BASS GATT service. #[pin_project(project = LocalServiceProj)] enum LocalServiceState<T: bt_gatt::ServerTypes> { @@ -78,6 +96,21 @@ fn is_published(&self) -> bool { matches!(self, LocalServiceState::Published { .. }) } + + fn service(&self) -> Option<&T::LocalService> { + match self { + Self::Published { service, .. } => Some(service), + _ => None, + } + } + + fn notify(&self, notification: &PendingNotification) { + let Some(service) = self.service() else { + warn!("Attempted to notify GATT peers on an unpublished service"); + return; + }; + service.notify(¬ification.handle, ¬ification.value, &[]); + } } impl<T: bt_gatt::ServerTypes> Stream for LocalServiceState<T> { @@ -335,6 +368,22 @@ Err(GattError::InsufficientResources) } + /// Allocates a new SourceId for the next available Receive State + /// characteristic entry. + /// + /// Returns the assigned [`Handle`] and [`SourceId`] on success. + fn allocate_receive_state_chrc(&mut self) -> Result<(Handle, SourceId), GattError> { + let handle = self + .receive_state_characteristics + .values() + .find(|c| c.state.is_empty()) + .map(|c| c.handle) + .ok_or(GattError::InsufficientResources)?; + + let source_id = self.allocate_source_id(handle)?; + Ok((handle, source_id)) + } + /// Handles a read request for a characteristic on the BASS server. fn handle_read(&self, handle: Handle, offset: usize, responder: impl ReadResponder) { // Only the Broadcast Receive State characteristics can be read. @@ -368,25 +417,14 @@ peer_id: PeerId, handle: Handle, value: &[u8], - responder: impl WriteResponder, - ) -> Option<ServerEvent> { + ) -> Result<ControlPointWriteResult, GattError> { // Only the Broadcast Audio Scan Control Point characteristic can be written to // by a client. See BASS v1.0 Section 3.3. if handle != CONTROL_POINT_HANDLE { - responder.error(GattError::WriteNotPermitted); - return None; + return Err(GattError::WriteNotPermitted); } - match self.handle_control_point_write(peer_id, value) { - Ok(event) => { - responder.acknowledge(); - Some(event) - } - Err(err) => { - responder.error(err); - None - } - } + self.handle_control_point_write(peer_id, value) } /// Returns true if the specified `source_id` is assigned to a non-empty @@ -402,45 +440,39 @@ &mut self, peer_id: PeerId, val_bytes: &[u8], - ) -> Result<ServerEvent, GattError> { + ) -> Result<ControlPointWriteResult, GattError> { let mut event = ServerEvent::decode(peer_id, val_bytes)?; - - match &mut event { - ServerEvent::RemoteScanState { .. } => {} + let notification = match &mut event { + ServerEvent::RemoteScanState { .. } => None, ServerEvent::AddSource { source_id, operation, .. } => { - *source_id = self.handle_add_source_write(operation)?; + let (notification, assigned_source_id) = self.handle_add_source_write(operation)?; + *source_id = assigned_source_id; + Some(notification) } ServerEvent::ModifySource { operation, .. } => { - self.handle_modify_source_write(operation)?; + let notification = self.handle_modify_source_write(operation)?; + Some(notification) } ServerEvent::SetBroadcastCode { source_id, .. } => { if !self.is_valid_source_id(*source_id) { return Err(ERROR_INVALID_SOURCE_ID); } + None } ServerEvent::RemoveSource { source_id, .. } => { - self.handle_remove_source_write(*source_id)?; + let notification = self.handle_remove_source_write(*source_id)?; + Some(notification) } - } + }; - Ok(event) + Ok(ControlPointWriteResult { event, notification }) } fn handle_add_source_write( &mut self, operation: &AddSourceOperation, - ) -> Result<SourceId, GattError> { - // Find the next available empty Receive State characteristic entry. - let handle = { - let Some(chrc) = - self.receive_state_characteristics.values().find(|c| c.state.is_empty()) - else { - return Err(GattError::InsufficientResources); - }; - chrc.handle - }; - - let source_id = self.allocate_source_id(handle)?; + ) -> Result<(PendingNotification, SourceId), GattError> { + let (handle, source_id) = self.allocate_receive_state_chrc()?; let chrc = self.receive_state_characteristics.get_mut(&handle).expect("just checked existence"); let pa_sync_state = PaSyncState::from_pa_sync(operation.pa_sync); @@ -461,21 +493,25 @@ subgroups, ); - chrc.state = BroadcastReceiveState::NonEmpty(initial_state); - Ok(source_id) + let new_state = BroadcastReceiveState::NonEmpty(initial_state); + let mut value = vec![0u8; new_state.encoded_len()]; + new_state.encode(&mut value).map_err(|_| GattError::UnlikelyError)?; + + chrc.state = new_state; + Ok((PendingNotification { handle, value }, source_id)) } fn handle_modify_source_write( &mut self, operation: &ModifySourceOperation, - ) -> Result<(), GattError> { - let Some(handle) = self.source_id_to_handle.get(&operation.source_id) else { + ) -> Result<PendingNotification, GattError> { + let Some(handle) = self.source_id_to_handle.get(&operation.source_id).copied() else { return Err(ERROR_INVALID_SOURCE_ID); }; - let Some(chrc) = self.receive_state_characteristics.get_mut(handle) else { + let Some(chrc) = self.receive_state_characteristics.get_mut(&handle) else { return Err(ERROR_INVALID_SOURCE_ID); }; - let BroadcastReceiveState::NonEmpty(ref mut state) = chrc.state else { + let BroadcastReceiveState::NonEmpty(ref state) = chrc.state else { return Err(ERROR_INVALID_SOURCE_ID); }; @@ -485,13 +521,22 @@ .map(|s| BigSubgroup::new(Some(s.bis_sync.clone())).with_metadata(s.metadata.clone())) .collect(); - state.subgroups = subgroups; - state.pa_sync_state = PaSyncState::from_pa_sync(operation.pa_sync); + let mut modified_state = state.clone(); + modified_state.subgroups = subgroups; + modified_state.pa_sync_state = PaSyncState::from_pa_sync(operation.pa_sync); - Ok(()) + let new_state = BroadcastReceiveState::NonEmpty(modified_state); + let mut value = vec![0u8; new_state.encoded_len()]; + new_state.encode(&mut value).map_err(|_| GattError::UnlikelyError)?; + + chrc.state = new_state; + Ok(PendingNotification { handle, value }) } - fn handle_remove_source_write(&mut self, source_id: SourceId) -> Result<(), GattError> { + fn handle_remove_source_write( + &mut self, + source_id: SourceId, + ) -> Result<PendingNotification, GattError> { let Some(handle) = self.source_id_to_handle.get(&source_id).copied() else { return Err(ERROR_INVALID_SOURCE_ID); }; @@ -504,7 +549,11 @@ let _ = self.source_id_to_handle.remove(&source_id); chrc.state = BroadcastReceiveState::Empty; - Ok(()) + + let mut value = vec![0u8; chrc.state.encoded_len()]; + chrc.state.encode(&mut value).map_err(|_| GattError::UnlikelyError)?; + + Ok(PendingNotification { handle, value }) } } @@ -545,6 +594,75 @@ Ok(()) } + /// Adds a new [`ReceiveState`] to the next available characteristic slot + /// and sends a GATT notification to all subscribed peers. + /// + /// Characteristic slots are allocated up-front when constructing the server + /// via [`ServerBuilder`]. This method populates an available empty slot + /// with `state` when initiated locally. + /// + /// Returns the assigned [`SourceId`] for the characteristic on success. + /// Returns [`Error::ServerFull`] if all characteristic slots are occupied. + pub fn add_receive_state(&mut self, mut state: ReceiveState) -> Result<SourceId, Error> { + let (_handle, source_id) = + self.state.allocate_receive_state_chrc().map_err(|_| Error::ServerFull)?; + state.source_id = source_id; + self.set_receive_state_and_notify(source_id, BroadcastReceiveState::NonEmpty(state))?; + Ok(source_id) + } + + /// Updates the existing [`BroadcastReceiveState`] characteristic with the + /// provided `state` and sends a GATT notification to all subscribed peers. + /// + /// Returns [`Error::InvalidSourceId`] if there is no such characteristic. + pub fn update_receive_state(&mut self, state: ReceiveState) -> Result<(), Error> { + self.set_receive_state_and_notify(state.source_id(), BroadcastReceiveState::NonEmpty(state)) + } + + /// Clears the entry for the specified Broadcast Receive State + /// characteristic and sends a GATT notification to all subscribed peers. + /// + /// Returns [`Error::InvalidSourceId`] if there is no such characteristic. + pub fn clear_receive_state(&mut self, source_id: SourceId) -> Result<(), Error> { + self.set_receive_state_and_notify(source_id, BroadcastReceiveState::Empty) + } + + /// Updates the state for the specified Broadcast Receive Characteristic and + /// sends a GATT notification to all subscribed peers. + /// + /// Returns [`Error::InvalidSourceId`] if there is no such characteristic. + fn set_receive_state_and_notify( + &mut self, + source_id: SourceId, + new_state: BroadcastReceiveState, + ) -> Result<(), Error> { + let handle = *self + .state + .source_id_to_handle + .get(&source_id) + .ok_or(Error::InvalidSourceId(source_id))?; + + let chrc = self + .state + .receive_state_characteristics + .get_mut(&handle) + .ok_or(Error::InvalidSourceId(source_id))?; + + let mut value = vec![0u8; new_state.encoded_len()]; + new_state.encode(&mut value)?; + + if new_state.is_empty() { + self.state.source_id_to_handle.remove(&source_id); + } else { + self.state.source_id_to_handle.insert(source_id, handle); + } + + chrc.state = new_state; + + self.local_service.notify(&PendingNotification { handle, value }); + + Ok(()) + } } impl<T: bt_gatt::ServerTypes> Stream for Server<T> { @@ -568,10 +686,17 @@ } ServiceEvent::Write { handle, peer_id, value, responder, .. } => { let val_vec = value.to_owned(); - if let Some(event) = - this.state.handle_write(peer_id, handle, &val_vec, responder) - { - return Poll::Ready(Some(Ok(event))); + match this.state.handle_write(peer_id, handle, &val_vec) { + Ok(result) => { + if let Some(notification) = result.notification { + this.local_service.notify(¬ification); + } + responder.acknowledge(); + return Poll::Ready(Some(Ok(result.event))); + } + Err(err) => { + responder.error(err); + } } } _ => { @@ -586,8 +711,11 @@ #[cfg(test)] mod tests { use super::*; + use assert_matches::assert_matches; use bt_bap::types::BroadcastId; + use bt_common::core::{AddressType, AdvertisingSetId, PeriodicAdvertisingInterval}; + use bt_common::generic_audio::metadata_ltv::Metadata; use bt_gatt::test_utils::{FakeServer, FakeServerEvent, FakeTypes}; use bt_gatt::types::GattError; use futures::task::Context; @@ -595,10 +723,10 @@ use futures::StreamExt; use std::task::Poll; - use crate::types::{EncryptionStatus, PaSync, PaSyncState, ReceiveState}; + use crate::types::{BigSubgroup, EncryptionStatus, PaSync, PaSyncState, ReceiveState}; - fn make_test_receive_state(source_id: SourceId) -> BroadcastReceiveState { - BroadcastReceiveState::NonEmpty(ReceiveState::new( + fn make_test_inner_receive_state(source_id: SourceId) -> ReceiveState { + ReceiveState::new( source_id, AddressType::Public, [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], @@ -607,7 +735,11 @@ PaSyncState::NotSynced, EncryptionStatus::NotEncrypted, vec![], - )) + ) + } + + fn make_test_receive_state(source_id: SourceId) -> BroadcastReceiveState { + BroadcastReceiveState::NonEmpty(make_test_inner_receive_state(source_id)) } #[test] @@ -888,13 +1020,22 @@ setup_test_server(vec![BroadcastReceiveState::Empty]); let mut event_stream = event_receiver.next(); + fake_gatt_server.incoming_client_configuration( + PeerId(1), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + + // 1. Remote client writes AddSource to Control Point (Handle 1) let add_op = AddSourceOperation::new( AddressType::Public, [1, 2, 3, 4, 5, 6], AdvertisingSetId::try_from(1).unwrap(), BroadcastId::try_from(0x123456).unwrap(), - PaSync::DoNotSync, - PeriodicAdvertisingInterval(0xFFFF), + PaSync::SyncPastAvailable, + PeriodicAdvertisingInterval(0x0010), vec![], ); let mut raw_bytes = vec![0u8; add_op.encoded_len()]; @@ -902,14 +1043,34 @@ fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes); + // Server stream emits AddSource event let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else { panic!("Expected ServerEvent"); }; - assert_eq!( - event, - ServerEvent::AddSource { peer_id: PeerId(1), source_id: 1, operation: add_op } - ); + let ServerEvent::AddSource { peer_id, source_id, operation } = event else { + panic!("Expected AddSource event, got {event:?}"); + }; + assert_eq!(peer_id, PeerId(1)); + assert_eq!(source_id, 1); + assert_eq!(operation, add_op); + // First notification: Initial minimal state created upon Control Point write + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected first Notified event"); + }; + assert_eq!(handle, Handle(2)); + let (first_notification_state, _) = BroadcastReceiveState::decode(&value); + let BroadcastReceiveState::NonEmpty(initial_state) = + first_notification_state.expect("valid decode") + else { + panic!("Expected NonEmpty state in first notification"); + }; + assert_eq!(initial_state.source_id, 1); + assert_eq!(initial_state.pa_sync_state, PaSyncState::SyncInfoRequest); + + // Write response is sent to client let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) = event_stream.poll_unpin(&mut noop_cx) else { @@ -917,6 +1078,30 @@ }; assert_eq!(handle, Handle(1)); assert!(value.is_ok()); + + // 2. Upper layer client processes the event, establishes PA sync, and updates + // state + let mut updated_state = initial_state; + updated_state.pa_sync_state = PaSyncState::Synced; + assert!(server.update_receive_state(updated_state).is_ok()); + + // Second notification: Upper layer client calls public API method + // (update_receive_state) + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, peers, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected second Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert_eq!(peers, vec![PeerId(1)]); + let (second_notification_state, _) = BroadcastReceiveState::decode(&value); + let BroadcastReceiveState::NonEmpty(final_state) = + second_notification_state.expect("valid decode") + else { + panic!("Expected NonEmpty state in second notification"); + }; + assert_eq!(final_state.source_id, 1); + assert_eq!(final_state.pa_sync_state, PaSyncState::Synced); } #[test] @@ -925,10 +1110,19 @@ setup_test_server(vec![make_test_receive_state(1)]); let mut event_stream = event_receiver.next(); + fake_gatt_server.incoming_client_configuration( + PeerId(1), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + + // 1. Remote client writes ModifySource to Control Point (Handle 1) let modify_op = ModifySourceOperation::new( 1, - PaSync::SyncPastAvailable, - PeriodicAdvertisingInterval(0x0010), + PaSync::DoNotSync, + PeriodicAdvertisingInterval(0xFFFF), vec![], ); let mut raw_bytes = vec![0u8; modify_op.encoded_len()]; @@ -936,6 +1130,7 @@ fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes); + // Server stream emits ModifySource event let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else { panic!("Expected ServerEvent"); }; @@ -944,6 +1139,20 @@ ServerEvent::ModifySource { peer_id: PeerId(1), source_id: 1, operation: modify_op } ); + // First notification: State updated upon Control Point write + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected first Notified event"); + }; + assert_eq!(handle, Handle(2)); + let (first_state, _) = BroadcastReceiveState::decode(&value); + let BroadcastReceiveState::NonEmpty(modified_state) = first_state.expect("valid decode") + else { + panic!("Expected NonEmpty state"); + }; + assert_eq!(modified_state.pa_sync_state, PaSyncState::NotSynced); + let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) = event_stream.poll_unpin(&mut noop_cx) else { @@ -951,6 +1160,27 @@ }; assert_eq!(handle, Handle(1)); assert!(value.is_ok()); + + // 2. Upper layer client processes the event and calls public API method + // (update_receive_state) + let mut updated_state = modified_state; + updated_state.pa_sync_state = PaSyncState::FailedToSync; + assert!(server.update_receive_state(updated_state).is_ok()); + + // Second notification: Upper layer client calls public API method + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, peers, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected second Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert_eq!(peers, vec![PeerId(1)]); + let (second_state, _) = BroadcastReceiveState::decode(&value); + let BroadcastReceiveState::NonEmpty(final_state) = second_state.expect("valid decode") + else { + panic!("Expected NonEmpty state"); + }; + assert_eq!(final_state.pa_sync_state, PaSyncState::FailedToSync); } #[test] @@ -993,6 +1223,14 @@ setup_test_server(vec![make_test_receive_state(1)]); let mut event_stream = event_receiver.next(); + fake_gatt_server.incoming_client_configuration( + PeerId(1), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + let remove_op = RemoveSourceOperation::new(1); let mut raw_bytes = vec![0u8; remove_op.encoded_len()]; remove_op.encode(&mut raw_bytes).expect("encode ok"); @@ -1004,6 +1242,14 @@ }; assert_eq!(event, ServerEvent::RemoveSource { peer_id: PeerId(1), source_id: 1 }); + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert!(value.is_empty()); + let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) = event_stream.poll_unpin(&mut noop_cx) else { @@ -1030,10 +1276,10 @@ panic!("Expected WriteResponded event"); }; assert_eq!(handle, Handle(2)); - assert!(matches!( + assert_matches!( value.unwrap_err(), bt_gatt::types::Error::Gatt(GattError::WriteNotPermitted) - )); + ); } #[test] @@ -1053,10 +1299,10 @@ panic!("Expected WriteResponded event"); }; assert_eq!(handle, Handle(1)); - assert!(matches!( + assert_matches!( value.unwrap_err(), bt_gatt::types::Error::Gatt(GattError::WriteRequestRejected) - )); + ); } #[test] @@ -1073,10 +1319,10 @@ panic!("Expected ReadResponded event"); }; assert_eq!(handle, CONTROL_POINT_HANDLE); - assert!(matches!( + assert_matches!( value.unwrap_err(), bt_gatt::types::Error::Gatt(GattError::ReadNotPermitted) - )); + ); } #[test] @@ -1093,10 +1339,7 @@ panic!("Expected ReadResponded event"); }; assert_eq!(handle, Handle(2)); - assert!(matches!( - value.unwrap_err(), - bt_gatt::types::Error::Gatt(GattError::InvalidOffset) - )); + assert_matches!(value.unwrap_err(), bt_gatt::types::Error::Gatt(GattError::InvalidOffset)); } #[test] @@ -1114,12 +1357,12 @@ assert!(server.publish(fake_gatt_server).is_ok()); - assert!(matches!( + assert_matches!( server.poll_next_unpin(&mut noop_cx), Poll::Ready(Some(Err(Error::Gatt(bt_gatt::types::Error::Gatt( GattError::UnlikelyError ))))) - )); + ); // Verifying server reset state to NotPublished so it is not published assert!(!server.is_published()); @@ -1146,6 +1389,212 @@ } #[test] + fn add_receive_state_success() { + let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) = + setup_test_server(vec![BroadcastReceiveState::Empty]); + let mut event_stream = event_receiver.next(); + + fake_gatt_server.incoming_client_configuration( + PeerId(10), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + + let state = make_test_inner_receive_state(99); + let result = server.add_receive_state(state); + assert_eq!(result.expect("add ok"), 1); + + // Expect the GATT notification for the updated characteristic. + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, peers, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert_eq!(value[0], 1); + assert_eq!(peers, vec![PeerId(10)]); + } + + #[test] + fn add_receive_state_when_full_fails() { + let (mut server, _fake_gatt_server, _event_receiver, _noop_cx) = + setup_test_server(vec![make_test_receive_state(1)]); + + let state = make_test_inner_receive_state(2); + let result = server.add_receive_state(state); + assert_matches!(result, Err(Error::ServerFull)); + } + + #[test] + fn update_receive_state_invalid_source_id_fails() { + let (mut server, _fake_gatt_server, _event_receiver, _noop_cx) = + setup_test_server(vec![BroadcastReceiveState::Empty]); + + let state = make_test_inner_receive_state(99); + let result = server.update_receive_state(state); + assert_matches!(result, Err(Error::InvalidSourceId(99))); + } + + #[test] + fn update_receive_state_notifies_peers() { + let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) = + setup_test_server(vec![make_test_receive_state(1)]); + let mut event_stream = event_receiver.next(); + + fake_gatt_server.incoming_client_configuration( + PeerId(10), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + + let new_inner = make_test_inner_receive_state(1); + let expected_state = BroadcastReceiveState::NonEmpty(new_inner.clone()); + let mut expected_bytes = vec![0u8; expected_state.encoded_len()]; + expected_state.encode(&mut expected_bytes).expect("encode ok"); + + let result = server.update_receive_state(new_inner); + assert!(result.is_ok()); + + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, peers, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert_eq!(value, expected_bytes); + assert_eq!(peers, vec![PeerId(10)]); + } + + #[test] + fn clear_receive_state_success() { + let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) = + setup_test_server(vec![make_test_receive_state(1)]); + let mut event_stream = event_receiver.next(); + + fake_gatt_server.incoming_client_configuration( + PeerId(10), + BASS_SERVICE_ID, + Handle(2), + bt_gatt::server::NotificationType::Notify, + ); + let _ = server.poll_next_unpin(&mut noop_cx); + + let result = server.clear_receive_state(1); + assert!(result.is_ok()); + + let Poll::Ready(Some(FakeServerEvent::Notified { handle, value, peers, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected Notified event"); + }; + assert_eq!(handle, Handle(2)); + assert!(value.is_empty()); + assert_eq!(peers, vec![PeerId(10)]); + } + + #[test] + fn clear_and_reuse_receive_state_slot() { + let (mut server, _fake_gatt_server, _event_receiver, _noop_cx) = + setup_test_server(vec![make_test_receive_state(1)]); + + // Clear existing state in slot 1 + assert!(server.clear_receive_state(1).is_ok()); + + // Adding a new state should successfully reuse the cleared slot and allocate + // the next ID + let new_state = make_test_inner_receive_state(50); + let result = server.add_receive_state(new_state); + assert_eq!(result.expect("add ok"), 2); + } + + #[test] + fn add_receive_state_sequential_slots() { + let (mut server, _fake_gatt_server, _event_receiver, _noop_cx) = setup_test_server(vec![ + BroadcastReceiveState::Empty, + BroadcastReceiveState::Empty, + BroadcastReceiveState::Empty, + ]); + + let state1 = make_test_inner_receive_state(10); + let state2 = make_test_inner_receive_state(20); + let state3 = make_test_inner_receive_state(30); + + assert_eq!(server.add_receive_state(state1).unwrap(), 1); + assert_eq!(server.add_receive_state(state2).unwrap(), 2); + assert_eq!(server.add_receive_state(state3).unwrap(), 3); + + // Fourth add should fail as server is full + let state4 = make_test_inner_receive_state(40); + assert_matches!(server.add_receive_state(state4), Err(Error::ServerFull)); + } + + #[test] + fn read_cleared_receive_state_returns_empty() { + let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) = + setup_test_server(vec![make_test_receive_state(1)]); + let mut event_stream = event_receiver.next(); + + // Clear the state + assert!(server.clear_receive_state(1).is_ok()); + + // Perform a GATT Read on Handle(2) + fake_gatt_server.incoming_read(PeerId(10), BASS_SERVICE_ID, Handle(2), 0); + let _ = server.poll_next_unpin(&mut noop_cx); + + let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected ReadResponded event"); + }; + assert_eq!(handle, Handle(2)); + let buf = value.expect("read ok"); + assert!(buf.is_empty()); + } + + #[test] + fn update_receive_state_encoding_failure_preserves_state() { + let initial_inner = make_test_inner_receive_state(1); + + let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) = + setup_test_server(vec![BroadcastReceiveState::NonEmpty(initial_inner.clone())]); + let mut event_stream = event_receiver.next(); + + // Create an invalid ReceiveState whose metadata total length exceeds 255 bytes + // (causing BigSubgroup encode to fail) + let vendor_metadata = vec![Metadata::AudioActiveState(true); 130]; + let mut invalid_inner = initial_inner.clone(); + invalid_inner.subgroups = vec![BigSubgroup::new(None).with_metadata(vendor_metadata)]; + + // Attempting to update to invalid_inner should return a Packet encoding error + let result = server.update_receive_state(invalid_inner); + assert_matches!(result, Err(Error::Packet(_))); + + // Perform a GATT Read on Handle(2) to verify internal memory state was NOT + // mutated to invalid_state + fake_gatt_server.incoming_read(PeerId(10), BASS_SERVICE_ID, Handle(2), 0); + let _ = server.poll_next_unpin(&mut noop_cx); + + let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) = + event_stream.poll_unpin(&mut noop_cx) + else { + panic!("Expected ReadResponded event"); + }; + assert_eq!(handle, Handle(2)); + + // Read value should successfully decode back to initial_inner state + let buf = value.expect("read ok"); + let (decoded_state, _) = BroadcastReceiveState::decode(&buf); + assert_eq!( + decoded_state.expect("decode ok"), + BroadcastReceiveState::NonEmpty(initial_inner) + ); + } + + #[test] fn allocate_source_id_skips_in_use_ids() { let mut state = ServerState::new(5); state.source_id_to_handle.insert(1, Handle(10));
diff --git a/rust/bt-bass/src/server/error.rs b/rust/bt-bass/src/server/error.rs index b5727f3..e0e4e7f 100644 --- a/rust/bt-bass/src/server/error.rs +++ b/rust/bt-bass/src/server/error.rs
@@ -23,9 +23,12 @@ #[error("An unsupported opcode ({0:#x}) used in Control Point operation")] OpCodeNotSupported(u8), - #[error("Invalid source id ({0}) used in Control Point operation")] + #[error("Invalid source id: {0}")] InvalidSourceId(SourceId), + #[error("All Broadcast Receive State characteristic slots are occupied")] + ServerFull, + #[error("Packet encoding/decoding error: {0}")] Packet(#[from] PacketError),
diff --git a/rust/bt-bass/src/types.rs b/rust/bt-bass/src/types.rs index 8b1d3cf..bd37c5a 100644 --- a/rust/bt-bass/src/types.rs +++ b/rust/bt-bass/src/types.rs
@@ -695,6 +695,13 @@ *self == BroadcastReceiveState::Empty } + pub fn source_id(&self) -> Option<SourceId> { + match self { + BroadcastReceiveState::Empty => None, + BroadcastReceiveState::NonEmpty(state) => Some(state.source_id), + } + } + pub fn broadcast_id(&self) -> Option<BroadcastId> { match self { BroadcastReceiveState::Empty => None,