blob: 691f15a3bc83bdc77d13cd50f466757bd4ccf021 [file]
// Copyright 2026 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::{BoxStream, StreamExt};
use std::collections::{HashMap, HashSet};
use bt_common::packet_encoding::Decodable;
use bt_common::Uuid;
use bt_gatt::client::{CharacteristicNotification, PeerService, ServiceCharacteristic};
use bt_gatt::types::Handle;
use crate::types::*;
/// 16-bit UUID value for the characteristics offered by the Audio
/// Stream Control Service.
pub const ASE_CONTROL_POINT_UUID: Uuid = Uuid::from_u16(0x2BC6);
pub const SINK_ASE_UUID: Uuid = Uuid::from_u16(0x2BC4);
pub const SOURCE_ASE_UUID: Uuid = Uuid::from_u16(0x2BC5);
#[derive(Debug, thiserror::Error, PartialEq, Clone)]
pub enum ClientError {
#[error("Remote server is missing control point characteristic")]
MissingControlPointCharacteristic,
#[error("Remote server should only have 1 control point characteristic")]
ExtraControlPointCharacteristic,
#[error("Remote server doesn't have any audio stream endpoints")]
MissingAudioStreamEndpoints,
#[error("Notification stream closed before receiving all responses")]
NotificationStreamClosed,
#[error(
"ASE {ase_id:?} transitioned to unexpected state: expected {expected:?}, got {actual:?}"
)]
UnexpectedStateTransition { ase_id: AseId, expected: AseState, actual: AseState },
#[error("Unknown ASE ID: {0:?}")]
UnknownAseId(AseId),
#[error("ASE {ase_id:?} is in invalid starting state {actual:?} for opcode {opcode:?}")]
InvalidStartState { ase_id: AseId, opcode: AseControlPointOpcode, actual: AseState },
}
/// Represents a single source/sink ASE state characteristic.
pub struct ClientEndpoint {
pub endpoint: AudioStreamEndpoint,
pub notification_stream:
BoxStream<'static, Result<CharacteristicNotification, bt_gatt::types::Error>>,
}
// See ASCS v1.0.1 4.2 for details.
pub struct AseControlPoint {
pub handle: Handle,
pub notification_stream:
BoxStream<'static, Result<CharacteristicNotification, bt_gatt::types::Error>>,
}
/// Creates Audio Stream Control Service (ASCS) client instance
pub struct AudioStreamControlServiceClient<T: bt_gatt::GattTypes> {
pub gatt_client: T::PeerService,
pub control_point: AseControlPoint,
pub endpoints: DiscoveredEndpoints,
}
pub struct DiscoveredEndpoints {
pub sink: HashMap<Handle, ClientEndpoint>,
pub source: HashMap<Handle, ClientEndpoint>,
}
impl DiscoveredEndpoints {
/// Returns the ASE IDs of all discovered sink Audio Stream Endpoints.
pub fn sink_ases(&self) -> Vec<AseId> {
self.sink.values().map(|e| e.endpoint.ase_id).collect()
}
/// Returns the ASE IDs of all discovered source Audio Stream Endpoints.
pub fn source_ases(&self) -> Vec<AseId> {
self.source.values().map(|e| e.endpoint.ase_id).collect()
}
pub(crate) fn lookup_by_ase_id(&self, ase_id: AseId) -> Option<&ClientEndpoint> {
if let Some(e) = self.sink.values().find(|e| e.endpoint.ase_id == ase_id) {
return Some(e);
}
if let Some(e) = self.source.values().find(|e| e.endpoint.ase_id == ase_id) {
return Some(e);
}
None
}
}
/// The outcome of an ASE control point operation, partitioning successfully
/// transitioned endpoints and failed response codes.
#[derive(Debug, Clone, PartialEq)]
pub struct AseControlOperationOutcome {
accepted: HashMap<AseId, AudioStreamEndpoint>,
rejected: HashMap<AseId, ResponseCode>,
}
impl AseControlOperationOutcome {
pub(crate) fn new(
accepted: HashMap<AseId, AudioStreamEndpoint>,
rejected: HashMap<AseId, ResponseCode>,
) -> Self {
Self { accepted, rejected }
}
/// Returns the map of successfully transitioned [`AudioStreamEndpoint`]s,
/// keyed by their [`AseId`].
pub fn accepted(&self) -> &HashMap<AseId, AudioStreamEndpoint> {
&self.accepted
}
/// Returns the map of rejected [`ResponseCode`]s, keyed by their [`AseId`].
pub fn rejected(&self) -> &HashMap<AseId, ResponseCode> {
&self.rejected
}
}
impl<T: bt_gatt::GattTypes> AudioStreamControlServiceClient<T> {
pub async fn create(gatt_client: T::PeerService) -> Result<Self, Error>
where
<T as bt_gatt::GattTypes>::NotificationStream: std::marker::Send,
{
let control_point = Self::discover_control_point(&gatt_client).await?;
let endpoints = Self::discover_all_endpoints(&gatt_client).await?;
Ok(Self { gatt_client, control_point, endpoints })
}
async fn discover_control_point(gatt_client: &T::PeerService) -> Result<AseControlPoint, Error>
where
<T as bt_gatt::GattTypes>::NotificationStream: std::marker::Send,
{
let cp_chars = ServiceCharacteristic::<T>::find(gatt_client, ASE_CONTROL_POINT_UUID)
.await
.map_err(Error::Gatt)?;
if cp_chars.is_empty() {
return Err(ClientError::MissingControlPointCharacteristic.into());
}
if cp_chars.len() > 1 {
return Err(ClientError::ExtraControlPointCharacteristic.into());
}
let cp_handle = *cp_chars[0].handle();
let cp_stream = gatt_client.subscribe(&cp_handle);
Ok(AseControlPoint { handle: cp_handle, notification_stream: cp_stream.boxed() })
}
async fn discover_all_endpoints(
gatt_client: &T::PeerService,
) -> Result<DiscoveredEndpoints, Error>
where
<T as bt_gatt::GattTypes>::NotificationStream: std::marker::Send,
{
let sink_chars = ServiceCharacteristic::<T>::find(gatt_client, SINK_ASE_UUID)
.await
.map_err(Error::Gatt)?;
let source_chars = ServiceCharacteristic::<T>::find(gatt_client, SOURCE_ASE_UUID)
.await
.map_err(Error::Gatt)?;
if sink_chars.is_empty() && source_chars.is_empty() {
return Err(ClientError::MissingAudioStreamEndpoints.into());
}
let mut sink_endpoints = HashMap::new();
for c in sink_chars {
let handle = *c.handle();
let endpoint =
Self::read_and_create_endpoint(gatt_client, handle, AudioDirection::Sink).await?;
let notification_stream = gatt_client.subscribe(&handle).boxed();
sink_endpoints.insert(handle, ClientEndpoint { endpoint, notification_stream });
}
let mut source_endpoints = HashMap::new();
for c in source_chars {
let handle = *c.handle();
let endpoint =
Self::read_and_create_endpoint(gatt_client, handle, AudioDirection::Source).await?;
let notification_stream = gatt_client.subscribe(&handle).boxed();
source_endpoints.insert(handle, ClientEndpoint { endpoint, notification_stream });
}
Ok(DiscoveredEndpoints { sink: sink_endpoints, source: source_endpoints })
}
async fn read_and_create_endpoint(
gatt_client: &T::PeerService,
handle: Handle,
direction: AudioDirection,
) -> Result<AudioStreamEndpoint, Error> {
let mut buf = vec![0; 255];
let (read_bytes, _truncated) =
gatt_client.read_characteristic(&handle, 0, &mut buf[..]).await.map_err(Error::Gatt)?;
let endpoint = AudioStreamEndpoint::from_char_value(handle, direction, &buf[0..read_bytes])
.map_err(|e| {
bt_gatt::types::Error::other(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Failed to decode AudioStreamEndpoint: {:?}", e),
))
})?;
Ok(endpoint)
}
fn verify_ase_ids<I>(&self, ase_ids: I) -> Result<(), Error>
where
I: IntoIterator<Item = AseId>,
{
for ase_id in ase_ids {
if self.endpoints.lookup_by_ase_id(ase_id).is_none() {
return Err(Error::Client(ClientError::UnknownAseId(ase_id)));
}
}
Ok(())
}
async fn read_and_update_endpoint(
&mut self,
handle: &Handle,
direction: AudioDirection,
) -> Result<AudioStreamEndpoint, Error> {
let mut buf = vec![0; 255];
let (read_bytes, _truncated) = self
.gatt_client
.read_characteristic(handle, 0, &mut buf[..])
.await
.map_err(Error::Gatt)?;
let endpoint =
AudioStreamEndpoint::from_char_value(*handle, direction, &buf[0..read_bytes]).map_err(
|e| {
bt_gatt::types::Error::other(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Failed to decode AudioStreamEndpoint: {:?}", e),
))
},
)?;
let endpoints = match direction {
AudioDirection::Sink => &mut self.endpoints.sink,
AudioDirection::Source => &mut self.endpoints.source,
};
if let Some(endpoint_handle) = endpoints.get_mut(handle) {
endpoint_handle.endpoint = endpoint.clone();
}
Ok(endpoint)
}
async fn write_operation(&mut self, operation: &AseControlOperation) -> Result<(), Error> {
use bt_common::packet_encoding::Encodable;
let mut buf = vec![0; operation.encoded_len()];
operation
.encode(&mut buf[..])
.map_err(|e| Error::Internal(format!("Failed to encode operation: {:?}", e)))?;
self.gatt_client
.write_characteristic(
&self.control_point.handle,
bt_gatt::types::WriteMode::WithoutResponse,
0,
&buf[..],
)
.await
.map_err(Error::Gatt)
}
async fn collect_operation_responses(
&mut self,
opcode: u8,
mut remaining_ases: HashSet<AseId>,
) -> Result<HashMap<AseId, ResponseCode>, Error> {
let mut responses = HashMap::new();
while !remaining_ases.is_empty() {
let notification = self
.control_point
.notification_stream
.next()
.await
.ok_or(Error::Client(ClientError::NotificationStreamClosed))?;
let notification = notification.map_err(Error::from)?;
let (cp_notification, _) = ControlPointNotification::decode(&notification.value[..]);
let cp_notification = cp_notification.map_err(|e| {
bt_gatt::types::Error::other(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Failed to decode ControlPointNotification: {:?}", e),
))
})?;
if cp_notification.opcode != opcode {
continue;
}
if cp_notification.number_of_ases == 0xFF {
return Err(Error::Internal(format!(
"Server rejected operation: {:?}",
cp_notification.responses
)));
}
for r in cp_notification.responses {
if remaining_ases.remove(&r.ase_id()) {
responses.insert(r.ase_id(), r);
}
}
}
Ok(responses)
}
async fn perform_and_verify_operation(
&mut self,
operation: AseControlOperation,
pending_ases: HashSet<AseId>,
) -> Result<AseControlOperationOutcome, Error> {
self.write_operation(&operation).await?;
let opcode_u8 = u8::try_from(&operation)?;
let responses = self.collect_operation_responses(opcode_u8, pending_ases).await?;
let mut successful_endpoints = HashMap::new();
let mut failed_responses = HashMap::new();
let opcode = AseControlPointOpcode::try_from(&operation).map_err(|e| {
Error::Internal(format!("Failed to resolve opcode from operation: {:?}", e))
})?;
for r in responses.into_values() {
if !r.is_success() {
failed_responses.insert(r.ase_id(), r);
continue;
}
let client_endpoint = self
.endpoints
.lookup_by_ase_id(r.ase_id())
.ok_or(Error::Client(ClientError::UnknownAseId(r.ase_id())))?;
let handle = client_endpoint.endpoint.handle;
let direction = client_endpoint.endpoint.direction;
let endpoint = self.read_and_update_endpoint(&handle, direction).await?;
if !opcode.allows_transition_to(endpoint.direction, &endpoint.state) {
let expected = opcode.operation_target_states(endpoint.direction)[0];
return Err(Error::Client(ClientError::UnexpectedStateTransition {
ase_id: endpoint.ase_id,
expected,
actual: endpoint.state,
}));
}
successful_endpoints.insert(endpoint.ase_id, endpoint);
}
Ok(AseControlOperationOutcome::new(successful_endpoints, failed_responses))
}
/// Performs the Configure Codec operation on one or more ASEs.
///
/// This method sends the Configure Codec request to the remote server,
/// waits for control point notifications for all involved ASEs, and
/// verifies that the successfully configured ASEs have transitioned to
/// the `CodecConfigured` state.
///
/// # Arguments
/// * `codec_configurations` - A vector of `CodecConfiguration` structs
/// containing the parameters for each ASE to be configured.
///
/// # Returns
/// On success, returns a tuple containing:
/// * A `HashMap` mapping each successfully configured `AseId` to its
/// supported `QosParameters` published by the server.
/// * A `Vec` containing the `ResponseCode` for any ASEs where the
/// configuration failed.
pub async fn configure_codec(
&mut self,
codec_configurations: Vec<CodecConfiguration>,
) -> Result<AseControlOperationOutcome, Error> {
let pending_ases: HashSet<AseId> = codec_configurations.iter().map(|c| c.ase_id).collect();
self.verify_ase_ids(pending_ases.iter().cloned())?;
let op = AseControlOperation::ConfigCodec { codec_configurations, responses: vec![] };
self.perform_and_verify_operation(op, pending_ases).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use bt_common::core::CodecId;
use bt_gatt::test_utils::{FakePeerService, FakeTypes};
use bt_gatt::types::{AttributePermissions, CharacteristicProperty};
use bt_gatt::Characteristic;
const CONTROL_POINT_HANDLE: Handle = Handle(1);
const SINK_ASE_HANDLE: Handle = Handle(2);
const SOURCE_ASE_HANDLE: Handle = Handle(3);
fn setup_fake_service() -> FakePeerService {
let mut service = FakePeerService::new();
// Add Control Point
service.add_characteristic(
Characteristic {
handle: CONTROL_POINT_HANDLE,
uuid: ASE_CONTROL_POINT_UUID,
properties: CharacteristicProperty::Write
| CharacteristicProperty::WriteWithoutResponse
| CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![],
);
// Add Sink ASE
service.add_characteristic(
Characteristic {
handle: SINK_ASE_HANDLE,
uuid: SINK_ASE_UUID,
properties: CharacteristicProperty::Read | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![0x01, 0x00], // ASE ID 1, State Idle
);
// Add Source ASE
service.add_characteristic(
Characteristic {
handle: SOURCE_ASE_HANDLE,
uuid: SOURCE_ASE_UUID,
properties: CharacteristicProperty::Read | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![0x02, 0x00], // ASE ID 2, State Idle
);
service
}
fn run_to_completion<F: std::future::Future>(fut: F) -> F::Output {
let mut fut = std::pin::pin!(fut);
let mut cx = futures::task::Context::from_waker(futures::task::noop_waker_ref());
match fut.as_mut().poll(&mut cx) {
std::task::Poll::Ready(res) => res,
std::task::Poll::Pending => panic!("Future did not complete synchronously"),
}
}
#[test]
fn client_creation_success() {
let service = setup_fake_service();
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service.clone());
let client = run_to_completion(client_fut).expect("client creation should succeed");
assert_eq!(client.control_point.handle, CONTROL_POINT_HANDLE);
assert_eq!(client.endpoints.sink.len(), 1);
assert_eq!(client.endpoints.source.len(), 1);
let sink = &client.endpoints.sink[&SINK_ASE_HANDLE].endpoint;
assert_eq!(sink.ase_id, AseId(1));
assert_eq!(sink.state, AseState::Idle);
assert_eq!(sink.direction, AudioDirection::Sink);
let source = &client.endpoints.source[&SOURCE_ASE_HANDLE].endpoint;
assert_eq!(source.ase_id, AseId(2));
assert_eq!(source.state, AseState::Idle);
assert_eq!(source.direction, AudioDirection::Source);
assert_eq!(client.endpoints.sink_ases(), vec![AseId(1)]);
assert_eq!(client.endpoints.source_ases(), vec![AseId(2)]);
}
#[test]
fn client_creation_missing_control_point() {
let mut service = FakePeerService::new();
// Only add Sink (no Control Point)
service.add_characteristic(
Characteristic {
handle: SINK_ASE_HANDLE,
uuid: SINK_ASE_UUID,
properties: CharacteristicProperty::Read | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![0x01, 0x00],
);
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service);
let err = match run_to_completion(client_fut) {
Err(e) => e,
Ok(_) => panic!("expected error, got Ok"),
};
assert!(matches!(err, Error::Client(ClientError::MissingControlPointCharacteristic)));
}
#[test]
fn client_creation_missing_endpoints() {
let mut service = FakePeerService::new();
// Only add Control Point (no endpoints)
service.add_characteristic(
Characteristic {
handle: CONTROL_POINT_HANDLE,
uuid: ASE_CONTROL_POINT_UUID,
properties: CharacteristicProperty::Write | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![],
);
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service);
let err = match run_to_completion(client_fut) {
Err(e) => e,
Ok(_) => panic!("expected error, got Ok"),
};
assert!(matches!(err, Error::Client(ClientError::MissingAudioStreamEndpoints)));
}
#[test]
fn client_creation_extra_control_point() {
let mut service = FakePeerService::new();
// Add first Control Point
service.add_characteristic(
Characteristic {
handle: CONTROL_POINT_HANDLE,
uuid: ASE_CONTROL_POINT_UUID,
properties: CharacteristicProperty::Write | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![],
);
// Add second Control Point
service.add_characteristic(
Characteristic {
handle: Handle(4),
uuid: ASE_CONTROL_POINT_UUID,
properties: CharacteristicProperty::Write | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![],
);
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service);
let err = match run_to_completion(client_fut) {
Err(e) => e,
Ok(_) => panic!("expected error, got Ok"),
};
assert!(matches!(err, Error::Client(ClientError::ExtraControlPointCharacteristic)));
}
#[test]
fn configure_codec() {
let mut service = setup_fake_service();
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service.clone());
let mut client = run_to_completion(client_fut).expect("client creation should succeed");
let config1 = CodecConfiguration {
ase_id: AseId(1),
target_latency: TargetLatency::TargetLowLatency,
target_phy: TargetPhy::Le1MPhy,
codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Lc3),
codec_specific_configuration: vec![],
};
let config2 = CodecConfiguration {
ase_id: AseId(2),
target_latency: TargetLatency::TargetLowLatency,
target_phy: TargetPhy::Le1MPhy,
codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Lc3),
codec_specific_configuration: vec![],
};
#[rustfmt::skip]
service.expect_characteristic_value(
&CONTROL_POINT_HANDLE,
vec![
0x01, // Opcode: Config Codec
0x02, // Num ASEs
0x01, 0x01, 0x01, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, // ASE 1 Config
0x02, 0x01, 0x01, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, // ASE 2 Config
],
);
#[rustfmt::skip]
service.notify(
&CONTROL_POINT_HANDLE,
Ok(CharacteristicNotification {
handle: CONTROL_POINT_HANDLE,
value: vec![
0x01, // Opcode
0x02, // Num ASEs
0x01, 0x00, 0x00, // ASE 1 Response: Success
0x02, 0x03, 0x02, // ASE 2 Response: Failed (Invalid ASE ID)
],
maybe_truncated: false,
}),
);
#[rustfmt::skip]
service.add_characteristic(
Characteristic {
handle: SINK_ASE_HANDLE,
uuid: SINK_ASE_UUID,
properties: CharacteristicProperty::Read | CharacteristicProperty::Notify,
permissions: AttributePermissions::default(),
descriptors: vec![],
},
vec![
0x01, // ASE ID: 1
0x01, // ASE State: Codec Configured
0x00, 0x01, 0x02, 0x0A, 0x00, 0x10, 0x27, 0x00, 0x40, 0x9C, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00,
],
);
let configure_fut = client.configure_codec(vec![config1, config2]);
let outcome = run_to_completion(configure_fut).expect("configure codec should succeed");
assert_eq!(
client.endpoints.sink[&SINK_ASE_HANDLE].endpoint.state,
AseState::CodecConfigured
);
assert_eq!(outcome.accepted().len(), 1);
let endpoint = &outcome.accepted()[&AseId(1)];
let qos = endpoint.qos_parameters().unwrap();
assert_eq!(qos.framing, Framing::Unframed);
assert_eq!(qos.preferred_phys, vec![Phy::Le1MPhy]);
assert_eq!(qos.preferred_retransmission_number, 2);
assert_eq!(outcome.rejected().len(), 1);
assert_eq!(outcome.rejected()[&AseId(2)], ResponseCode::InvalidAseId { value: 2 });
}
#[test]
fn configure_codec_fail_unknown_ase_id() {
let service = setup_fake_service();
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service.clone());
let mut client = run_to_completion(client_fut).expect("client creation should succeed");
let config = CodecConfiguration {
ase_id: AseId(99), // Non-existent ASE ID
target_latency: TargetLatency::TargetLowLatency,
target_phy: TargetPhy::Le1MPhy,
codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Lc3),
codec_specific_configuration: vec![],
};
let configure_fut = client.configure_codec(vec![config]);
let err = run_to_completion(configure_fut).expect_err("should fail due to unknown ASE ID");
assert!(matches!(err, Error::Client(ClientError::UnknownAseId(AseId(99)))));
}
#[test]
fn configure_codec_global_error() {
let mut service = setup_fake_service();
let client_fut = AudioStreamControlServiceClient::<FakeTypes>::create(service.clone());
let mut client = run_to_completion(client_fut).expect("client creation should succeed");
let config = CodecConfiguration {
ase_id: AseId(1),
target_latency: TargetLatency::TargetLowLatency,
target_phy: TargetPhy::Le1MPhy,
codec_id: CodecId::Assigned(bt_common::core::CodingFormat::Lc3),
codec_specific_configuration: vec![],
};
#[rustfmt::skip]
service.expect_characteristic_value(
&CONTROL_POINT_HANDLE,
vec![
0x01, // Opcode: Config Codec
0x01, // Num ASEs
0x01, 0x01, 0x01, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, // ASE 1 Config
],
);
// Let's mimic server side throwing a global invalid length error.
// (in reality this wouldn't happen because our configure codec operation was
// encoded correctly).
#[rustfmt::skip]
service.notify(
&CONTROL_POINT_HANDLE,
Ok(CharacteristicNotification {
handle: CONTROL_POINT_HANDLE,
value: vec![
0x01, // Opcode
0xFF, // Num ASEs: Global Error
0x00, 0x02, 0x00, // ASE 0, Invalid Length, None
],
maybe_truncated: false,
}),
);
let configure_fut = client.configure_codec(vec![config]);
let err = run_to_completion(configure_fut).expect_err("should fail with global error");
assert!(matches!(
err,
Error::Internal(ref s) if s.contains("Server rejected operation") && s.contains("InvalidLength")
));
}
}