bt_hfp/audio/
inband.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::format_err;
6use fidl_fuchsia_bluetooth_bredr as bredr;
7use fidl_fuchsia_media as media;
8use fuchsia_async as fasync;
9use fuchsia_audio_codec::{StreamProcessor, StreamProcessorOutputStream};
10use fuchsia_audio_device::stream_config::SoftStreamConfig;
11use fuchsia_audio_device::{AudioFrameSink, AudioFrameStream, AudioStreamItem};
12use fuchsia_bluetooth::types::{PeerId, peer_audio_stream_id};
13use fuchsia_sync::Mutex;
14use futures::stream::BoxStream;
15use futures::task::Context;
16use futures::{AsyncWriteExt, FutureExt, StreamExt};
17use log::{error, info, warn};
18use media::AudioDeviceEnumeratorProxy;
19use std::pin::pin;
20
21use crate::audio::{Control, ControlEvent, Error, HF_INPUT_UUID, HF_OUTPUT_UUID};
22use crate::codec_id::CodecId;
23use crate::sco;
24
25/// Audio Control for inband audio, i.e. encoding and decoding audio before sending them
26/// to the controller via HCI (in contrast to offloaded audio).
27pub struct InbandControl {
28    audio_core: media::AudioDeviceEnumeratorProxy,
29    session_task: Option<(PeerId, fasync::Task<()>)>,
30    event_sender: Mutex<futures::channel::mpsc::Sender<ControlEvent>>,
31    stream: Mutex<Option<futures::channel::mpsc::Receiver<ControlEvent>>>,
32}
33
34// Setup for a running AudioSession.
35// AudioSesison::run() consumes the session and should handle the data path in both directions:
36//   - SCO -> decoder -> audio_core input (audio_frame_sink)
37//   - audio_core output -> encoder -> SCO
38struct AudioSession {
39    audio_frame_sink: AudioFrameSink,
40    audio_frame_stream: AudioFrameStream,
41    sco: sco::Connection,
42    codec: CodecId,
43    decoder: StreamProcessor,
44    encoder: StreamProcessor,
45    event_sender: futures::channel::mpsc::Sender<ControlEvent>,
46}
47
48impl AudioSession {
49    fn setup(
50        connection: sco::Connection,
51        codec: CodecId,
52        audio_frame_sink: AudioFrameSink,
53        audio_frame_stream: AudioFrameStream,
54        event_sender: futures::channel::mpsc::Sender<ControlEvent>,
55    ) -> Result<Self, Error> {
56        if !codec.is_supported() {
57            return Err(Error::UnsupportedParameters {
58                source: format_err!("unsupported codec {codec}"),
59            });
60        }
61        let decoder = StreamProcessor::create_decoder(codec.mime_type()?, Some(codec.oob_bytes()))
62            .map_err(|e| Error::audio_core(format_err!("creating decoder: {e:?}")))?;
63        let encoder = StreamProcessor::create_encoder(codec.try_into()?, codec.try_into()?)
64            .map_err(|e| Error::audio_core(format_err!("creating encoder: {e:?}")))?;
65        Ok(Self {
66            sco: connection,
67            decoder,
68            encoder,
69            audio_frame_sink,
70            audio_frame_stream,
71            codec,
72            event_sender,
73        })
74    }
75
76    async fn encoder_to_sco(
77        mut encoded_stream: StreamProcessorOutputStream,
78        proxy: bredr::ScoConnectionProxy,
79        codec: CodecId,
80    ) -> Error {
81        // Pre-allocate the packet vector and reuse to avoid allocating for every packet.
82        let packet: Vec<u8> = vec![0; 60]; // SCO has 60 byte packets
83        let mut request =
84            bredr::ScoConnectionWriteRequest { data: Some(packet), ..Default::default() };
85
86        const MSBC_ENCODED_LEN: usize = 57; // Length of a MSBC packet after encoding.
87        if codec == CodecId::MSBC {
88            let packet: &mut [u8] = request.data.as_mut().unwrap().as_mut_slice();
89            packet[0] = 0x01; // H2 header has a constant part (0b1000_0000_0001_AABB) with AABB
90            // cycling 0000, 0011, 1100, 1111
91        }
92        // The H2 Header marker cycle, with the constant part
93        let mut h2_marker = [0x08u8, 0x38, 0xc8, 0xf8].iter().cycle();
94        loop {
95            match encoded_stream.next().await {
96                Some(Ok(encoded)) => {
97                    if codec == CodecId::MSBC {
98                        if encoded.len() % MSBC_ENCODED_LEN != 0 {
99                            warn!("Got {} bytes, uneven number of packets", encoded.len());
100                        }
101                        for sbc_packet in encoded.as_slice().chunks_exact(MSBC_ENCODED_LEN) {
102                            let packet: &mut [u8] = request.data.as_mut().unwrap().as_mut_slice();
103                            packet[1] = *h2_marker.next().unwrap();
104                            packet[2..59].copy_from_slice(sbc_packet);
105                            if let Err(e) = proxy.write(&request).await {
106                                return e.into();
107                            }
108                        }
109                    } else {
110                        // CVSD has no padding or header. Encoder sends us multiples of 60 bytes as
111                        // long as we provide a multiple of 7.5ms audio packets.
112                        for cvsd_packet in encoded.as_slice().chunks_exact(60) {
113                            let packet: &mut [u8] = request.data.as_mut().unwrap().as_mut_slice();
114                            packet.copy_from_slice(cvsd_packet);
115                            if let Err(e) = proxy.write(&request).await {
116                                return e.into();
117                            }
118                        }
119                    }
120                }
121                Some(Err(e)) => {
122                    warn!("Error in encoding: {e:?}");
123                    return Error::audio_core(format_err!("Couldn't read encoded: {e:?}"));
124                }
125                None => {
126                    warn!("Error in encoding: Stream is ended!");
127                    return Error::audio_core(format_err!("Encoder stream ended early"));
128                }
129            }
130        }
131    }
132
133    async fn pcm_to_encoder(mut encoder: StreamProcessor, mut stream: AudioFrameStream) -> Error {
134        loop {
135            match stream.next().await {
136                Some(Ok(AudioStreamItem::Data(pcm))) => {
137                    if let Err(e) = encoder.write_all(pcm.as_slice()).await {
138                        return Error::audio_core(format_err!("write to encoder: {e:?}"));
139                    }
140                    // Packets should be exactly the right size.
141                    if let Err(e) = encoder.flush().await {
142                        return Error::audio_core(format_err!("flush encoder: {e:?}"));
143                    }
144                }
145                Some(Ok(AudioStreamItem::AudioDisabled)) => {
146                    continue;
147                }
148                Some(Err(e)) => {
149                    warn!("Audio output error: {e:?}");
150                    return Error::audio_core(format_err!("output error: {e:?}"));
151                }
152                None => {
153                    warn!("Ran out of audio input!");
154                    return Error::audio_core(format_err!("Audio input end"));
155                }
156            }
157        }
158    }
159
160    async fn decoder_to_pcm(
161        mut decoded_stream: StreamProcessorOutputStream,
162        mut sink: AudioFrameSink,
163    ) -> Error {
164        let mut decoded_packets = 0;
165        loop {
166            match decoded_stream.next().await {
167                Some(Ok(decoded)) => {
168                    decoded_packets += 1;
169                    if decoded_packets % 500 == 0 {
170                        info!(
171                            "Got {} decoded bytes from decoder: {decoded_packets} packets",
172                            decoded.len()
173                        );
174                    }
175                    if let Err(e) = sink.write_all(decoded.as_slice()).await {
176                        warn!("Error sending to sink: {e:?}");
177                        return Error::audio_core(format_err!("send to sink: {e:?}"));
178                    }
179                }
180                Some(Err(e)) => {
181                    warn!("Error in decoding: {e:?}");
182                    return Error::audio_core(format_err!("Couldn't read decoder: {e:?}"));
183                }
184                None => {
185                    warn!("Error in decoding: Stream is ended!");
186                    return Error::audio_core(format_err!("Decoder stream ended early"));
187                }
188            }
189        }
190    }
191
192    async fn sco_to_decoder(
193        proxy: bredr::ScoConnectionProxy,
194        mut decoder: StreamProcessor,
195        codec: CodecId,
196    ) -> Error {
197        loop {
198            let data = match proxy.read().await {
199                Ok(bredr::ScoConnectionReadResponse { data: Some(data), .. }) => data,
200                Ok(_) => return Error::audio_core(format_err!("Invalid Read response")),
201                Err(e) => return e.into(),
202            };
203            let packet = match codec {
204                CodecId::CVSD => data.as_slice(),
205                CodecId::MSBC => {
206                    // H2 Header (two octets) is present on packets when WBS is used
207                    let (_header, packet) = data.as_slice().split_at(2);
208                    if packet[0] != 0xad {
209                        info!(
210                            "Packet didn't start with syncword: {:#02x} {}",
211                            packet[0],
212                            packet.len()
213                        );
214                    }
215                    packet
216                }
217                _ => {
218                    return Error::UnsupportedParameters {
219                        source: format_err!("Unknown CodecId: {codec:?}"),
220                    };
221                }
222            };
223            if let Err(e) = decoder.write_all(packet).await {
224                return Error::audio_core(format_err!("Failed to write to decoder: {e:?}"));
225            }
226            // TODO(https://fxbug.dev/42073275): buffer some packets before flushing instead of flushing on
227            // every one.
228            if let Err(e) = decoder.flush().await {
229                return Error::audio_core(format_err!("Failed to flush decoder: {e:?}"));
230            }
231        }
232    }
233
234    async fn run(mut self) {
235        let peer_id = self.sco.peer_id;
236        let Ok(encoded_stream) = self.encoder.take_output_stream() else {
237            error!("Couldn't take encoder output stream");
238            return;
239        };
240        let sco_write =
241            AudioSession::encoder_to_sco(encoded_stream, self.sco.proxy.clone(), self.codec);
242        let sco_write = pin!(sco_write);
243        let audio_to_encoder = AudioSession::pcm_to_encoder(self.encoder, self.audio_frame_stream);
244        let audio_to_encoder = pin!(audio_to_encoder);
245
246        let Ok(decoded_stream) = self.decoder.take_output_stream() else {
247            error!("Couldn't take decoder output stream");
248            return;
249        };
250        let decoder_to_sink =
251            pin!(AudioSession::decoder_to_pcm(decoded_stream, self.audio_frame_sink));
252        let sco_read =
253            AudioSession::sco_to_decoder(self.sco.proxy.clone(), self.decoder, self.codec);
254        let sco_read = pin!(sco_read);
255        let e = futures::select! {
256            e = audio_to_encoder.fuse() => { warn!(e:?; "PCM to encoder write"); e},
257            e = sco_write.fuse() => { warn!(e:?; "Write encoded to SCO"); e},
258            e = sco_read.fuse() => { warn!(e:?; "SCO read to decoder"); e},
259            e = decoder_to_sink.fuse() => { warn!(e:?; "SCO decoder to PCM"); e},
260        };
261        let _ = self.event_sender.try_send(ControlEvent::Stopped { id: peer_id, error: Some(e) });
262    }
263
264    fn start(self) -> fasync::Task<()> {
265        fasync::Task::spawn(self.run())
266    }
267}
268
269impl InbandControl {
270    pub fn create(proxy: AudioDeviceEnumeratorProxy) -> Result<Self, Error> {
271        let (sender, receiver) = futures::channel::mpsc::channel(1);
272        Ok(Self {
273            audio_core: proxy,
274            session_task: None,
275            event_sender: Mutex::new(sender),
276            stream: Mutex::new(Some(receiver)),
277        })
278    }
279
280    fn running_id(&mut self) -> Option<PeerId> {
281        self.session_task
282            .as_mut()
283            .and_then(|(running, task)| {
284                let mut cx = Context::from_waker(&std::task::Waker::noop());
285                // We are the only thing that polls this task, so we are ok to poll it and throw away a
286                // wake.
287                task.poll_unpin(&mut cx).is_pending().then_some(running)
288            })
289            .copied()
290    }
291
292    const LOCAL_MONOTONIC_CLOCK_DOMAIN: u32 = 0;
293
294    // This is currently 2x an SCO frame which holds 7.5ms
295    // This must be a multiple of 7.5ms for the CVSD encoder to not have any remainder bytes.
296    const AUDIO_BUFFER_DURATION: zx::MonotonicDuration = zx::MonotonicDuration::from_millis(15);
297
298    fn start_input(&mut self, peer_id: PeerId, codec_id: CodecId) -> Result<AudioFrameSink, Error> {
299        let audio_dev_id = peer_audio_stream_id(peer_id, HF_INPUT_UUID);
300        let (client, sink) = SoftStreamConfig::create_input(
301            &audio_dev_id,
302            "Fuchsia",
303            super::DEVICE_NAME,
304            Self::LOCAL_MONOTONIC_CLOCK_DOMAIN,
305            codec_id.try_into()?,
306            Self::AUDIO_BUFFER_DURATION,
307        )
308        .map_err(|e| Error::audio_core(format_err!("Couldn't create input: {e:?}")))?;
309
310        self.audio_core.add_device_by_channel(super::DEVICE_NAME, true, client)?;
311        Ok(sink)
312    }
313
314    fn start_output(
315        &mut self,
316        peer_id: PeerId,
317        codec_id: CodecId,
318    ) -> Result<AudioFrameStream, Error> {
319        let audio_dev_id = peer_audio_stream_id(peer_id, HF_OUTPUT_UUID);
320        let (client, stream) = SoftStreamConfig::create_output(
321            &audio_dev_id,
322            "Fuchsia",
323            super::DEVICE_NAME,
324            Self::LOCAL_MONOTONIC_CLOCK_DOMAIN,
325            codec_id.try_into()?,
326            Self::AUDIO_BUFFER_DURATION,
327            zx::MonotonicDuration::from_millis(0),
328        )
329        .map_err(|e| Error::audio_core(format_err!("Couldn't create output: {e:?}")))?;
330        self.audio_core.add_device_by_channel(super::DEVICE_NAME, false, client)?;
331        Ok(stream)
332    }
333}
334
335impl Control for InbandControl {
336    fn start(
337        &mut self,
338        id: PeerId,
339        connection: sco::Connection,
340        codec: CodecId,
341    ) -> Result<(), Error> {
342        if let Some(running) = self.running_id() {
343            if running == id {
344                return Err(Error::AlreadyStarted);
345            }
346            return Err(Error::UnsupportedParameters {
347                source: format_err!("Only one peer can be started inband at once"),
348            });
349        }
350        let frame_sink = self.start_input(id, codec)?;
351        let frame_stream = self.start_output(id, codec)?;
352        let session = AudioSession::setup(
353            connection,
354            codec,
355            frame_sink,
356            frame_stream,
357            self.event_sender.lock().clone(),
358        )?;
359        self.session_task = Some((id, session.start()));
360        Ok(())
361    }
362
363    fn stop(&mut self, id: PeerId) -> Result<(), Error> {
364        if self.running_id() != Some(id) {
365            return Err(Error::NotStarted);
366        }
367        self.session_task = None;
368        let _ = self.event_sender.get_mut().try_send(ControlEvent::Stopped { id, error: None });
369        Ok(())
370    }
371
372    fn connect(&mut self, _id: PeerId, _supported_codecs: &[CodecId]) {
373        // Nothing to do here
374    }
375
376    fn disconnect(&mut self, id: PeerId) {
377        let _ = self.stop(id);
378    }
379
380    fn take_events(&self) -> BoxStream<'static, ControlEvent> {
381        self.stream.lock().take().unwrap().boxed()
382    }
383
384    fn failed_request(&self, _request: ControlEvent, _error: Error) {
385        // We send no requests, so ignore this.
386    }
387}
388
389#[cfg(test)]
390mod tests {
391    use super::*;
392
393    use fidl_fuchsia_bluetooth_bredr::ScoConnectionRequestStream;
394
395    use crate::sco::test_utils::connection_for_codec;
396
397    /// A "Zero input response" SBC packet.  This is what SBC encodes to (with the MSBC settings)
398    /// when passed a flat input at zero.  Each packet represents 7.5 milliseconds of audio.
399    const ZERO_INPUT_SBC_PACKET: [u8; 60] = [
400        0x80, 0x10, 0xad, 0x00, 0x00, 0xc5, 0x00, 0x00, 0x00, 0x00, 0x77, 0x6d, 0xb6, 0xdd, 0xdb,
401        0x6d, 0xb7, 0x76, 0xdb, 0x6d, 0xdd, 0xb6, 0xdb, 0x77, 0x6d, 0xb6, 0xdd, 0xdb, 0x6d, 0xb7,
402        0x76, 0xdb, 0x6d, 0xdd, 0xb6, 0xdb, 0x77, 0x6d, 0xb6, 0xdd, 0xdb, 0x6d, 0xb7, 0x76, 0xdb,
403        0x6d, 0xdd, 0xb6, 0xdb, 0x77, 0x6d, 0xb6, 0xdd, 0xdb, 0x6d, 0xb7, 0x76, 0xdb, 0x6c, 0x00,
404    ];
405
406    /// A "zero input response" CVSD packet.
407    const ZERO_INPUT_CVSD_PACKET: [u8; 60] = [0x55; 60];
408
409    #[derive(PartialEq, Debug)]
410    enum ProcessedRequest {
411        ScoRead,
412        ScoWrite(Vec<u8>),
413    }
414
415    // Processes one sco request.  Returns true if the stream was ended.
416    async fn process_sco_request(
417        sco_request_stream: &mut ScoConnectionRequestStream,
418        read_data: Vec<u8>,
419    ) -> Option<ProcessedRequest> {
420        match sco_request_stream.next().await {
421            Some(Ok(bredr::ScoConnectionRequest::Read { responder })) => {
422                let response = bredr::ScoConnectionReadResponse {
423                    status_flag: Some(bredr::RxPacketStatus::CorrectlyReceivedData),
424                    data: Some(read_data),
425                    ..Default::default()
426                };
427                responder.send(&response).expect("sends okay");
428                Some(ProcessedRequest::ScoRead)
429            }
430            Some(Ok(bredr::ScoConnectionRequest::Write { payload, responder })) => {
431                responder.send().expect("response to write");
432                Some(ProcessedRequest::ScoWrite(payload.data.unwrap()))
433            }
434            None => None,
435            x => panic!("Expected read or write requests, got {x:?}"),
436        }
437    }
438
439    #[fuchsia::test]
440    async fn reads_audio_from_connection() {
441        let (proxy, _audio_enumerator_requests) =
442            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
443        let mut control = InbandControl::create(proxy).unwrap();
444
445        let (connection, mut sco_request_stream) =
446            connection_for_codec(PeerId(1), CodecId::MSBC, true);
447
448        control.start(PeerId(1), connection, CodecId::MSBC).expect("should be able to start");
449
450        let (connection2, _request_stream) = connection_for_codec(PeerId(1), CodecId::MSBC, true);
451        let _ = control
452            .start(PeerId(1), connection2, CodecId::MSBC)
453            .expect_err("Starting twice shouldn't be allowed");
454
455        // Test note: 10 packets is not enough to force a write to audio, which will stall this test if
456        // it's not started.
457        for _ in 1..10 {
458            assert_eq!(
459                Some(ProcessedRequest::ScoRead),
460                process_sco_request(&mut sco_request_stream, ZERO_INPUT_SBC_PACKET.to_vec()).await
461            );
462        }
463
464        control.stop(PeerId(1)).expect("should be able to stop");
465        let _ = control.stop(PeerId(1)).expect_err("can't stop a stopped thing");
466
467        // Should be able to drain the requests.
468        let mut extra_requests = 0;
469        while let Some(r) =
470            process_sco_request(&mut sco_request_stream, ZERO_INPUT_SBC_PACKET.to_vec()).await
471        {
472            assert_eq!(ProcessedRequest::ScoRead, r);
473            extra_requests += 1;
474        }
475
476        info!("Got {extra_requests} extra ScoConnectionProxy Requests after stop");
477    }
478
479    #[fuchsia::test]
480    async fn audio_setup_error_bad_codec() {
481        let (proxy, _) =
482            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
483        let mut control = InbandControl::create(proxy).unwrap();
484
485        let (connection, _sco_request_stream) =
486            connection_for_codec(PeerId(1), CodecId::MSBC, true);
487        let res = control.start(PeerId(1), connection, 0xD0u8.into());
488        assert!(res.is_err());
489    }
490
491    #[fuchsia::test]
492    async fn decode_sco_audio_path() {
493        use fidl_fuchsia_hardware_audio as audio;
494        let (proxy, mut audio_enumerator_requests) =
495            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
496        let mut control = InbandControl::create(proxy).unwrap();
497
498        let (connection, mut sco_request_stream) =
499            connection_for_codec(PeerId(1), CodecId::MSBC, true);
500
501        control.start(PeerId(1), connection, CodecId::MSBC).expect("should be able to start");
502
503        let audio_input_stream_config;
504        let mut _audio_output_stream_config;
505        loop {
506            match audio_enumerator_requests.next().await {
507                Some(Ok(media::AudioDeviceEnumeratorRequest::AddDeviceByChannel {
508                    is_input,
509                    channel,
510                    ..
511                })) => {
512                    if is_input {
513                        audio_input_stream_config = channel.into_proxy();
514                        break;
515                    } else {
516                        _audio_output_stream_config = channel.into_proxy();
517                    }
518                }
519                x => panic!("Expected audio device by channel, got {x:?}"),
520            }
521        }
522
523        let (ring_buffer, server) = fidl::endpoints::create_proxy::<audio::RingBufferMarker>();
524        audio_input_stream_config
525            .create_ring_buffer(&CodecId::MSBC.try_into().unwrap(), server)
526            .expect("create ring buffer");
527
528        // We need to write to the stream at least once to start it up.
529        assert_eq!(
530            Some(ProcessedRequest::ScoRead),
531            process_sco_request(&mut sco_request_stream, ZERO_INPUT_SBC_PACKET.to_vec()).await
532        );
533
534        let notifications_per_ring = 20;
535        // Request a 1-second audio buffer. This is guaranteed to be greater than 1 second, since
536        // the driver must reserve any space it needs ON TOP OF the client-requested 16000 bytes.
537        let (frames, _vmo) = ring_buffer
538            .get_vmo(16000, notifications_per_ring)
539            .await
540            .expect("fidl")
541            .expect("response");
542
543        // To be deterministic, we set the first notification before even starting the ring-buffer.
544        let mut position_info = ring_buffer.watch_clock_recovery_position_info();
545        let mut position_notifications = 0;
546
547        let _ = ring_buffer.start().await;
548
549        // For 100 MSBC Audio frames, we get 7.5 x 100 = 750 milliseconds, or 12000 frames.
550        let frames_per_notification = frames / notifications_per_ring;
551        // As noted above, `frames` > 16000, so `frames_per_notification` > 800. Assuming the ring-
552        // buffer is < 17000 frames, `expected_notifications` will be 14.xx (as u32: 14), not 15.
553        let expected_notifications = 12000 / frames_per_notification;
554
555        // We might receive the first notification as early as ring-buffer position 0,
556        // so we check for a notification before processing the first chunk of data.
557        if position_info.poll_unpin(&mut Context::from_waker(&std::task::Waker::noop())).is_ready()
558        {
559            position_notifications += 1;
560            position_info = ring_buffer.watch_clock_recovery_position_info();
561        }
562        for _ in 1..100 {
563            assert_eq!(
564                Some(ProcessedRequest::ScoRead),
565                process_sco_request(&mut sco_request_stream, ZERO_INPUT_SBC_PACKET.to_vec()).await
566            );
567            // We are the only ones polling position_info, so we can ignore wakeups (noop waker).
568            if position_info
569                .poll_unpin(&mut Context::from_waker(&std::task::Waker::noop()))
570                .is_ready()
571            {
572                position_notifications += 1;
573                position_info = ring_buffer.watch_clock_recovery_position_info();
574            }
575        }
576
577        // The audio driver protocol require notification VALUES [timestamp, position] to tightly
578        // correlate. It is less concerned with notification ARRIVAL TIMES; these could occur up to
579        // 1 notification's duration early. Thus, if we expect X notifications, then we allow X+1.
580        assert!(position_notifications >= expected_notifications);
581        assert!(position_notifications <= expected_notifications + 1);
582    }
583
584    #[fuchsia::test]
585    async fn encode_sco_audio_path_msbc() {
586        use fidl_fuchsia_hardware_audio as audio;
587        let (proxy, mut audio_enumerator_requests) =
588            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
589        let mut control = InbandControl::create(proxy).unwrap();
590
591        let (connection, mut sco_request_stream) =
592            connection_for_codec(PeerId(1), CodecId::MSBC, true);
593
594        control.start(PeerId(1), connection, CodecId::MSBC).expect("should be able to start");
595
596        let audio_output_stream_config;
597        let mut _audio_input_stream_config;
598        loop {
599            match audio_enumerator_requests.next().await {
600                Some(Ok(media::AudioDeviceEnumeratorRequest::AddDeviceByChannel {
601                    is_input,
602                    channel,
603                    ..
604                })) => {
605                    if !is_input {
606                        audio_output_stream_config = channel.into_proxy();
607                        break;
608                    } else {
609                        _audio_input_stream_config = channel.into_proxy();
610                    }
611                }
612                x => panic!("Expected audio device by channel, got {x:?}"),
613            }
614        }
615
616        let (ring_buffer, server) = fidl::endpoints::create_proxy::<audio::RingBufferMarker>();
617        audio_output_stream_config
618            .create_ring_buffer(&CodecId::MSBC.try_into().unwrap(), server)
619            .unwrap();
620
621        // Note: we don't need to read from the stream to start it, it gets polled automatically by
622        // the read task.
623
624        let notifications_per_ring = 20;
625        // Request at least 1 second of audio buffer.
626        let (_frames, _vmo) = ring_buffer
627            .get_vmo(16000, notifications_per_ring)
628            .await
629            .expect("fidl")
630            .expect("response");
631
632        let _ = ring_buffer.start().await;
633
634        // Expect 100 MSBC Audio frames, which should take ~ 750 milliseconds.
635        let next_header = &mut [0x01, 0x08];
636        for _sco_frame in 1..100 {
637            'sco: loop {
638                match process_sco_request(&mut sco_request_stream, ZERO_INPUT_SBC_PACKET.to_vec())
639                    .await
640                {
641                    Some(ProcessedRequest::ScoRead) => continue 'sco,
642                    Some(ProcessedRequest::ScoWrite(data)) => {
643                        assert_eq!(60, data.len());
644                        // Skip the H2 header which changes for every packet.
645                        assert_eq!(&ZERO_INPUT_SBC_PACKET[2..], &data[2..]);
646                        assert_eq!(next_header, &data[0..2]);
647                        // Prep for the next heade
648                        match next_header[1] {
649                            0x08 => next_header[1] = 0x38,
650                            0x38 => next_header[1] = 0xc8,
651                            0xc8 => next_header[1] = 0xf8,
652                            0xf8 => next_header[1] = 0x08,
653                            _ => unreachable!(),
654                        };
655                        break 'sco;
656                    }
657                    x => panic!("Expected read or write but got {x:?}"),
658                };
659            }
660        }
661    }
662
663    #[fuchsia::test]
664    async fn encode_sco_audio_path_cvsd() {
665        use fidl_fuchsia_hardware_audio as audio;
666        let (proxy, mut audio_enumerator_requests) =
667            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
668        let mut control = InbandControl::create(proxy).unwrap();
669
670        let (connection, mut sco_request_stream) =
671            connection_for_codec(PeerId(1), CodecId::CVSD, true);
672
673        control.start(PeerId(1), connection, CodecId::CVSD).expect("should be able to start");
674
675        let audio_output_stream_config;
676        let mut _audio_input_stream_config;
677        loop {
678            match audio_enumerator_requests.next().await {
679                Some(Ok(media::AudioDeviceEnumeratorRequest::AddDeviceByChannel {
680                    is_input,
681                    channel,
682                    ..
683                })) => {
684                    if !is_input {
685                        audio_output_stream_config = channel.into_proxy();
686                        break;
687                    } else {
688                        _audio_input_stream_config = channel.into_proxy();
689                    }
690                }
691                x => panic!("Expected audio device by channel, got {x:?}"),
692            }
693        }
694
695        let (ring_buffer, server) = fidl::endpoints::create_proxy::<audio::RingBufferMarker>();
696        audio_output_stream_config
697            .create_ring_buffer(&CodecId::CVSD.try_into().unwrap(), server)
698            .unwrap();
699
700        // Note: we don't need to read from the stream to start it, it gets polled automatically by
701        // the read task.
702
703        let notifications_per_ring = 10;
704        // Request at least 1 second of audio buffer.
705        let (_frames, _vmo) = ring_buffer
706            .get_vmo(64000, notifications_per_ring)
707            .await
708            .expect("fidl")
709            .expect("response");
710
711        let _ = ring_buffer.start().await;
712
713        // Expect 100 CVSD Audio frames, which should take ~ 750 milliseconds.
714        for _sco_frame in 1..100 {
715            'sco: loop {
716                match process_sco_request(&mut sco_request_stream, ZERO_INPUT_CVSD_PACKET.to_vec())
717                    .await
718                {
719                    Some(ProcessedRequest::ScoRead) => continue 'sco,
720                    Some(ProcessedRequest::ScoWrite(data)) => {
721                        // Confirm the data is right
722                        assert_eq!(60, data.len());
723                        assert_eq!(&ZERO_INPUT_CVSD_PACKET, data.as_slice());
724                        break 'sco;
725                    }
726                    x => panic!("Expected read or write but got {x:?}"),
727                };
728            }
729        }
730    }
731
732    #[fuchsia::test]
733    async fn read_from_audio_output() {
734        use fidl_fuchsia_hardware_audio as audio;
735        let (proxy, mut audio_enumerator_requests) =
736            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
737        let mut control = InbandControl::create(proxy).unwrap();
738
739        let (connection, mut sco_request_stream) =
740            connection_for_codec(PeerId(1), CodecId::MSBC, true);
741
742        control.start(PeerId(1), connection, CodecId::MSBC).expect("should be able to start");
743
744        let audio_output_stream_config;
745        let mut _audio_input_stream_config;
746        loop {
747            match audio_enumerator_requests.next().await {
748                Some(Ok(media::AudioDeviceEnumeratorRequest::AddDeviceByChannel {
749                    is_input,
750                    channel,
751                    ..
752                })) => {
753                    if !is_input {
754                        audio_output_stream_config = channel.into_proxy();
755                        break;
756                    } else {
757                        _audio_input_stream_config = channel.into_proxy();
758                    }
759                }
760                x => panic!("Expected audio device by channel, got {x:?}"),
761            }
762        }
763
764        let (ring_buffer, server) = fidl::endpoints::create_proxy::<audio::RingBufferMarker>();
765        audio_output_stream_config
766            .create_ring_buffer(&CodecId::MSBC.try_into().unwrap(), server)
767            .expect("create ring buffer");
768
769        let notifications_per_ring = 20;
770        // Request at least 1 second of audio buffer.
771        let (_frames, _vmo) = ring_buffer
772            .get_vmo(16000, notifications_per_ring)
773            .await
774            .expect("fidl")
775            .expect("response");
776
777        let _ = ring_buffer.start().await;
778
779        // We should be just reading from the audio output, track via position notifications.
780        // 20 position notifications happen in one second.
781        'position_notifications: for i in 1..20 {
782            let mut position_info = ring_buffer.watch_clock_recovery_position_info();
783            loop {
784                let sco_activity = Box::pin(process_sco_request(
785                    &mut sco_request_stream,
786                    ZERO_INPUT_SBC_PACKET.to_vec(),
787                ));
788                use futures::future::Either;
789                match futures::future::select(position_info, sco_activity).await {
790                    Either::Left((result, _sco_fut)) => {
791                        assert!(result.is_ok(), "Position Info failed at {i}");
792                        continue 'position_notifications;
793                    }
794                    Either::Right((_sco_pkt, position_info_fut)) => {
795                        position_info = position_info_fut;
796                    }
797                }
798            }
799        }
800    }
801
802    #[fuchsia::test]
803    async fn audio_output_error_sends_to_events() {
804        let (proxy, mut audio_enumerator_requests) =
805            fidl::endpoints::create_proxy_and_stream::<media::AudioDeviceEnumeratorMarker>();
806        let mut control = InbandControl::create(proxy).unwrap();
807        let mut events = control.take_events();
808
809        let (connection, _sco_request_stream) =
810            connection_for_codec(PeerId(1), CodecId::MSBC, true);
811
812        control.start(PeerId(1), connection, CodecId::MSBC).expect("should be able to start");
813
814        let audio_output_stream_config;
815        let mut _audio_input_stream_config;
816        loop {
817            match audio_enumerator_requests.next().await {
818                Some(Ok(media::AudioDeviceEnumeratorRequest::AddDeviceByChannel {
819                    is_input,
820                    channel,
821                    ..
822                })) => {
823                    if !is_input {
824                        audio_output_stream_config = channel.into_proxy();
825                        break;
826                    } else {
827                        _audio_input_stream_config = channel.into_proxy();
828                    }
829                }
830                x => panic!("Expected audio device by channel, got {x:?}"),
831            }
832        }
833
834        drop(audio_output_stream_config);
835
836        // Events should produce an error because there was an issue with audio output.
837        match events.next().await {
838            Some(ControlEvent::Stopped { id, error: Some(_) }) => {
839                assert_eq!(PeerId(1), id);
840            }
841            x => panic!("Expected the peer to have error stop, but got {x:?}"),
842        };
843    }
844}
bt_hfp/audio/inband.rs

bt_hfp/audio/
inband.rs
