bt_a2dp/
stream.rs

1// Copyright 2020 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::Error;
6use bt_avdtp::{
7    self as avdtp, ErrorCode, ServiceCapability, ServiceCategory, StreamEndpoint, StreamEndpointId,
8};
9use fidl_fuchsia_bluetooth_bredr::AudioOffloadExtProxy;
10use fuchsia_bluetooth::types::PeerId;
11use fuchsia_inspect::{self as inspect, Property};
12use fuchsia_inspect_derive::{AttachError, Inspect};
13use futures::future::BoxFuture;
14use futures::{FutureExt, TryFutureExt};
15use log::{info, warn};
16use std::collections::HashMap;
17use std::fmt;
18use std::sync::Arc;
19use std::time::Duration;
20
21use crate::codec::{CodecNegotiation, MediaCodecConfig};
22use crate::media_task::{MediaTask, MediaTaskBuilder, MediaTaskError, MediaTaskRunner};
23
24/// Manages a local StreamEndpoint and its associated media task, starting and stopping the
25/// related media task in sync with the endpoint's configured or streaming state.
26/// Note that this does not coordinate state with peer, which is done by bt_a2dp::Peer.
27pub struct Stream {
28    endpoint: StreamEndpoint,
29    /// The builder for media tasks associated with this endpoint.
30    media_task_builder: Arc<Box<dyn MediaTaskBuilder>>,
31    /// The MediaTaskRunner for this endpoint, if it is configured.
32    media_task_runner: Option<Box<dyn MediaTaskRunner>>,
33    /// The MediaTask, if it is running.
34    media_task: Option<Box<dyn MediaTask>>,
35    /// The peer associated with this endpoint, if it is configured.
36    /// Used during reconfiguration for MediaTask recreation.
37    peer_id: Option<PeerId>,
38    /// Inspect Node for this stream
39    inspect: fuchsia_inspect::Node,
40}
41
42impl fmt::Debug for Stream {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        f.debug_struct("Stream")
45            .field("endpoint", &self.endpoint)
46            .field("peer_id", &self.peer_id)
47            .field("has media_task", &self.media_task.is_some())
48            .finish()
49    }
50}
51
52impl Inspect for &mut Stream {
53    // Set up the StreamEndpoint to update the state
54    // The MediaTask node will be created when the media task is started.
55    fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> {
56        self.inspect = parent.create_child(name.as_ref());
57
58        let endpoint_state_prop = self.inspect.create_string("endpoint_state", "");
59        let callback =
60            move |stream: &StreamEndpoint| endpoint_state_prop.set(&format!("{:?}", stream));
61        callback(self.endpoint());
62        self.endpoint_mut().set_update_callback(Some(Box::new(callback)));
63        Ok(())
64    }
65}
66
67impl Stream {
68    pub fn build(endpoint: StreamEndpoint, media_task_builder: Box<dyn MediaTaskBuilder>) -> Self {
69        Self {
70            endpoint,
71            media_task_builder: Arc::new(media_task_builder),
72            media_task_runner: None,
73            media_task: None,
74            peer_id: None,
75            inspect: Default::default(),
76        }
77    }
78
79    fn as_new(&self) -> Self {
80        Self {
81            endpoint: self.endpoint.as_new(),
82            media_task_builder: self.media_task_builder.clone(),
83            media_task_runner: None,
84            media_task: None,
85            peer_id: None,
86            inspect: Default::default(),
87        }
88    }
89
90    pub fn endpoint(&self) -> &StreamEndpoint {
91        &self.endpoint
92    }
93
94    pub fn endpoint_mut(&mut self) -> &mut StreamEndpoint {
95        &mut self.endpoint
96    }
97
98    fn media_codec_config(&self) -> Option<MediaCodecConfig> {
99        find_codec_capability(self.endpoint.capabilities())
100            .and_then(|x| MediaCodecConfig::try_from(x).ok())
101    }
102
103    /// Returns true if the config given is a supported configuration of this stream
104    /// Used when the stream is being configured to a specific configuration
105    fn config_supported(&self, config: &MediaCodecConfig) -> bool {
106        let Some(supported) = self.media_codec_config() else {
107            return false;
108        };
109        supported.supports(&config)
110    }
111
112    /// Returns true if this stream and the given config are compatible - a valid configuration
113    /// of this stream can be found within the capabilities of the given config.
114    fn config_compatible(&self, config: &MediaCodecConfig) -> bool {
115        let Some(supported) = self.media_codec_config() else {
116            return false;
117        };
118        MediaCodecConfig::negotiate(&supported, config).is_some()
119    }
120
121    fn build_media_task(
122        &self,
123        peer_id: &PeerId,
124        config: &MediaCodecConfig,
125    ) -> Option<Box<dyn MediaTaskRunner>> {
126        match self.media_task_builder.configure(peer_id, &config) {
127            Err(e) => {
128                warn!("Failed to build media task: {e:?}");
129                None
130            }
131            Ok(mut media_task_runner) => {
132                if let Err(e) = media_task_runner.iattach(&self.inspect, "media_task") {
133                    info!("Media Task inspect: {e}");
134                }
135                Some(media_task_runner)
136            }
137        }
138    }
139
140    fn supported_config_from_capability(
141        &self,
142        requested_cap: &ServiceCapability,
143    ) -> Option<MediaCodecConfig> {
144        MediaCodecConfig::try_from(requested_cap).ok().filter(|c| self.config_supported(c))
145    }
146
147    pub fn configure(
148        &mut self,
149        peer_id: &PeerId,
150        remote_id: &StreamEndpointId,
151        capabilities: Vec<ServiceCapability>,
152    ) -> Result<(), (ServiceCategory, ErrorCode)> {
153        if self.media_task.is_some() {
154            return Err((ServiceCategory::None, ErrorCode::BadState));
155        }
156        let unsupported = ErrorCode::UnsupportedConfiguration;
157        let codec_cap =
158            find_codec_capability(&capabilities).ok_or((ServiceCategory::None, unsupported))?;
159        let media_unsupported = (ServiceCategory::MediaCodec, unsupported);
160        let config = self.supported_config_from_capability(codec_cap).ok_or(media_unsupported)?;
161        self.media_task_runner =
162            Some(self.build_media_task(peer_id, &config).ok_or(media_unsupported)?);
163        self.peer_id = Some(peer_id.clone());
164        self.endpoint.configure(remote_id, capabilities)
165    }
166
167    pub fn set_delay(&mut self, delay: Duration) -> Result<(), ErrorCode> {
168        let Some(runner) = self.media_task_runner.as_mut() else {
169            return Err(ErrorCode::BadState);
170        };
171        match runner.set_delay(delay) {
172            Err(MediaTaskError::NotSupported) => Err(ErrorCode::NotSupportedCommand),
173            Err(_) => Err(ErrorCode::BadState),
174            Ok(()) => Ok(()),
175        }
176    }
177
178    pub fn reconfigure(
179        &mut self,
180        capabilities: Vec<ServiceCapability>,
181    ) -> Result<(), (ServiceCategory, ErrorCode)> {
182        let bad_state = (ServiceCategory::None, ErrorCode::BadState);
183        let _peer_id = self.peer_id.as_ref().ok_or(bad_state)?;
184        if let Some(requested_codec_cap) = find_codec_capability(&capabilities) {
185            let unsupported = (ServiceCategory::MediaCodec, ErrorCode::UnsupportedConfiguration);
186            let requested =
187                self.supported_config_from_capability(requested_codec_cap).ok_or(unsupported)?;
188            self.media_task_runner
189                .as_mut()
190                .ok_or(bad_state)?
191                .reconfigure(&requested)
192                .or(Err(unsupported))?;
193        }
194        self.endpoint.reconfigure(capabilities)
195    }
196
197    fn media_runner_ref(&mut self) -> Result<&mut Box<dyn MediaTaskRunner>, ErrorCode> {
198        self.media_task_runner.as_mut().ok_or(ErrorCode::BadState)
199    }
200
201    /// Attempt to start the endpoint.
202    /// If the endpoint is successfully started, the media task is started and a future that
203    /// will finish when the media task finishes is returned.
204    pub fn start(&mut self) -> Result<BoxFuture<'static, Result<(), Error>>, ErrorCode> {
205        if self.media_task_runner.is_none() {
206            return Err(ErrorCode::BadState);
207        };
208        let transport = self.endpoint.take_transport().ok_or(ErrorCode::BadState)?;
209        let _ = self.endpoint.start()?;
210        let offload = self.endpoint.audio_offload();
211        let mut task = match self.media_runner_ref()?.start(transport, offload) {
212            Ok(media_task) => media_task,
213            Err(_e) => {
214                let _ = self.endpoint.suspend()?;
215                return Err(ErrorCode::BadState);
216            }
217        };
218        let finished = task.finished();
219        self.media_task = Some(task);
220        Ok(finished.err_into().boxed())
221    }
222
223    /// Suspends the media processor and endpoint.
224    pub fn suspend(&mut self) -> Result<(), ErrorCode> {
225        self.endpoint.suspend()?;
226        let _ = self.media_task.take().ok_or(ErrorCode::BadState)?.stop();
227        Ok(())
228    }
229
230    fn stop_media_task(&mut self) {
231        if let Some(mut task) = self.media_task.take() {
232            // Ignoring stop errors, best effort.
233            let _ = task.stop();
234        }
235        self.media_task_runner = None;
236        self.peer_id = None;
237    }
238
239    /// Releases the endpoint and stops the processing of audio.
240    pub fn release(
241        &mut self,
242        responder: avdtp::SimpleResponder,
243        peer: &avdtp::Peer,
244    ) -> avdtp::Result<()> {
245        self.stop_media_task();
246        self.endpoint.release(responder, peer)
247    }
248
249    pub fn abort(&mut self) {
250        self.stop_media_task();
251        self.endpoint.abort()
252    }
253
254    pub async fn initiate_abort(&mut self, peer: &avdtp::Peer) {
255        self.stop_media_task();
256        self.endpoint.initiate_abort(peer).await
257    }
258}
259
260fn find_codec_capability(capabilities: &[ServiceCapability]) -> Option<&ServiceCapability> {
261    capabilities.iter().find(|cap| cap.category() == ServiceCategory::MediaCodec)
262}
263
264/// Iterator which generates SEIDs.  Used by StreamsBuilder to get valid SEIDs.
265#[derive(Clone, Debug)]
266struct SeidRangeFrom {
267    from: u8,
268}
269
270impl Default for SeidRangeFrom {
271    fn default() -> Self {
272        Self { from: 1 }
273    }
274}
275
276impl Iterator for SeidRangeFrom {
277    type Item = u8;
278
279    fn next(&mut self) -> Option<Self::Item> {
280        let res = self.from;
281        if self.from == 0x3E {
282            self.from = 0x01;
283        } else {
284            self.from += 1;
285        }
286        Some(res)
287    }
288}
289
290/// Builds a set of streams, based on the capabilities of a set of MediaTaskBuilders that are
291/// supported and configured by the system.
292pub struct StreamsBuilder {
293    builders: Vec<Box<dyn MediaTaskBuilder>>,
294    seid_range: SeidRangeFrom,
295    node: inspect::Node,
296}
297
298impl Default for StreamsBuilder {
299    fn default() -> Self {
300        Self {
301            builders: Default::default(),
302            seid_range: SeidRangeFrom { from: Self::START_SEID },
303            node: Default::default(),
304        }
305    }
306}
307
308impl Clone for StreamsBuilder {
309    fn clone(&self) -> Self {
310        Self {
311            builders: self.builders.clone(),
312            node: Default::default(),
313            seid_range: self.seid_range.clone(),
314        }
315    }
316}
317
318impl StreamsBuilder {
319    // Randomly chosen by fair dice roll
320    // TODO(https://fxbug.dev/337321738): Do better for randomizing this maybe
321    const START_SEID: u8 = 8;
322
323    /// Add a builder to the set of builders used to generate streams.
324    pub fn add_builder(&mut self, builder: impl MediaTaskBuilder + 'static) {
325        self.builders.push(Box::new(builder));
326        self.node.record_uint("builders", self.builders.len() as u64);
327    }
328
329    pub async fn peer_streams(
330        &self,
331        peer_id: &PeerId,
332        offload: Option<AudioOffloadExtProxy>,
333    ) -> Result<Streams, MediaTaskError> {
334        let mut streams = Streams::default();
335        let mut seid_range = self.seid_range.clone();
336        for builder in &self.builders {
337            let endpoint_type = builder.direction();
338            let supported_res = builder.supported_configs(peer_id, offload.clone()).await;
339            let Ok(supported) = supported_res else {
340                info!(e:? = supported_res.err().unwrap(); "Failed to get supported configs from builder, skipping");
341                continue;
342            };
343            let codec_caps = supported.iter().map(ServiceCapability::from);
344            for codec_cap in codec_caps {
345                let capabilities = match endpoint_type {
346                    avdtp::EndpointType::Source => vec![
347                        ServiceCapability::MediaTransport,
348                        ServiceCapability::DelayReporting,
349                        codec_cap,
350                    ],
351                    avdtp::EndpointType::Sink => {
352                        vec![ServiceCapability::MediaTransport, codec_cap]
353                    }
354                };
355                let endpoint = avdtp::StreamEndpoint::new(
356                    seid_range.next().unwrap(),
357                    avdtp::MediaType::Audio,
358                    endpoint_type,
359                    capabilities,
360                )?;
361                streams.insert(Stream::build(endpoint, builder.clone()));
362            }
363        }
364        Ok(streams)
365    }
366
367    pub async fn negotiation(
368        &self,
369        peer_id: &PeerId,
370        offload: Option<AudioOffloadExtProxy>,
371        preferred_direction: avdtp::EndpointType,
372    ) -> Result<CodecNegotiation, Error> {
373        let mut caps_available = Vec::new();
374        for builder in &self.builders {
375            caps_available.extend(
376                builder
377                    .supported_configs(peer_id, offload.clone())
378                    .await?
379                    .iter()
380                    .map(ServiceCapability::from),
381            );
382        }
383        Ok(CodecNegotiation::build(caps_available, preferred_direction)?)
384    }
385}
386
387impl Inspect for &mut StreamsBuilder {
388    fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> {
389        self.node = parent.create_child(name.as_ref());
390        self.node.record_uint("builders", self.builders.len() as u64);
391        Ok(())
392    }
393}
394
395/// A set of streams, indexed by their local endpoint ID.
396#[derive(Default)]
397pub struct Streams {
398    streams: HashMap<StreamEndpointId, Stream>,
399    inspect_node: fuchsia_inspect::Node,
400}
401
402impl fmt::Debug for Streams {
403    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
404        f.debug_struct("Streams").field("streams", &self.streams).finish()
405    }
406}
407
408impl Streams {
409    /// Makes a copy of this set of streams, but with all streams copied with their states set to
410    /// idle.
411    pub fn as_new(&self) -> Self {
412        let streams =
413            self.streams.iter().map(|(id, stream)| (id.clone(), stream.as_new())).collect();
414        Self { streams, ..Default::default() }
415    }
416
417    /// Returns true if there are no streams in the set.
418    pub fn is_empty(&self) -> bool {
419        self.streams.is_empty()
420    }
421
422    /// Inserts a stream, indexing it by the local endpoint id.
423    /// It replaces any other stream with the same endpoint id.
424    pub fn insert(&mut self, stream: Stream) {
425        let local_id = stream.endpoint().local_id().clone();
426        if self.streams.insert(local_id.clone(), stream).is_some() {
427            warn!("Replacing stream with local id {local_id}");
428        }
429    }
430
431    /// Retrieves a reference to the Stream referenced by `id`, if the stream exists,
432    pub fn get(&self, id: &StreamEndpointId) -> Option<&Stream> {
433        self.streams.get(id)
434    }
435
436    /// Retrieves a mutable reference to the Stream referenced by `id`, if the stream exists,
437    pub fn get_mut(&mut self, id: &StreamEndpointId) -> Option<&mut Stream> {
438        self.streams.get_mut(id)
439    }
440
441    /// Returns a vector of information on all the contained streams.
442    pub fn information(&self) -> Vec<avdtp::StreamInformation> {
443        self.streams.values().map(|x| x.endpoint().information()).collect()
444    }
445
446    /// Returns streams that are in the open (established but not streaming) state
447    pub fn open(&self) -> impl Iterator<Item = &Stream> {
448        self.streams.values().filter(|s| s.endpoint().state() == avdtp::StreamState::Open)
449    }
450
451    /// Returns streams that are streaming.
452    pub fn streaming(&self) -> impl Iterator<Item = &Stream> {
453        self.streams.values().filter(|s| s.endpoint().state() == avdtp::StreamState::Streaming)
454    }
455
456    /// Finds streams in the set which are compatible with `codec_config`.
457    pub fn compatible(&self, codec_config: MediaCodecConfig) -> impl Iterator<Item = &Stream> {
458        self.streams.values().filter(move |s| s.config_compatible(&codec_config))
459    }
460}
461
462impl Inspect for &mut Streams {
463    // Attach self to `parent`
464    fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> {
465        self.inspect_node = parent.create_child(name.as_ref());
466        for stream in self.streams.values_mut() {
467            stream.iattach(&self.inspect_node, inspect::unique_name("stream_"))?;
468        }
469        Ok(())
470    }
471}
472
473#[cfg(test)]
474pub(crate) mod tests {
475    use super::*;
476    use bt_channel_test_support::{Transport, create_test_channels};
477    use fuchsia_async as fasync;
478    use std::pin::pin;
479    use std::task::Poll;
480    use test_case::test_case;
481
482    use crate::media_task::tests::TestMediaTaskBuilder;
483    use crate::media_types::*;
484
485    pub(crate) fn sbc_mediacodec_capability() -> avdtp::ServiceCapability {
486        let sbc_codec_info = SbcCodecInfo::new(
487            SbcSamplingFrequency::FREQ48000HZ,
488            SbcChannelMode::MONO | SbcChannelMode::JOINT_STEREO,
489            SbcBlockCount::MANDATORY_SRC,
490            SbcSubBands::MANDATORY_SRC,
491            SbcAllocation::MANDATORY_SRC,
492            SbcCodecInfo::BITPOOL_MIN,
493            SbcCodecInfo::BITPOOL_MAX,
494        )
495        .expect("SBC codec info");
496
497        ServiceCapability::MediaCodec {
498            media_type: avdtp::MediaType::Audio,
499            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
500            codec_extra: sbc_codec_info.to_bytes().to_vec(),
501        }
502    }
503
504    pub(crate) fn aac_mediacodec_capability(bitrate: u32) -> avdtp::ServiceCapability {
505        let codec_info = AacCodecInfo::new(
506            AacObjectType::MANDATORY_SRC,
507            AacSamplingFrequency::FREQ48000HZ,
508            AacChannels::TWO,
509            true,
510            bitrate,
511        )
512        .expect("should work");
513        ServiceCapability::MediaCodec {
514            media_type: avdtp::MediaType::Audio,
515            codec_type: avdtp::MediaCodecType::AUDIO_AAC,
516            codec_extra: codec_info.to_bytes().to_vec(),
517        }
518    }
519
520    pub(crate) fn make_sbc_endpoint(seid: u8, direction: avdtp::EndpointType) -> StreamEndpoint {
521        StreamEndpoint::new(
522            seid,
523            avdtp::MediaType::Audio,
524            direction,
525            vec![avdtp::ServiceCapability::MediaTransport, sbc_mediacodec_capability()],
526        )
527        .expect("endpoint creation should succeed")
528    }
529
530    const LOW_BITRATE: u32 = 320_000;
531    const HIGH_BITRATE: u32 = 393_216;
532
533    pub(crate) fn make_aac_endpoint(seid: u8, direction: avdtp::EndpointType) -> StreamEndpoint {
534        StreamEndpoint::new(
535            seid,
536            avdtp::MediaType::Audio,
537            direction,
538            vec![avdtp::ServiceCapability::MediaTransport, aac_mediacodec_capability(LOW_BITRATE)],
539        )
540        .expect("endpoint creation should succeed")
541    }
542
543    fn make_stream(seid: u8, codec_type: avdtp::MediaCodecType) -> Stream {
544        let endpoint = match codec_type {
545            avdtp::MediaCodecType::AUDIO_SBC => {
546                make_sbc_endpoint(seid, avdtp::EndpointType::Source)
547            }
548            avdtp::MediaCodecType::AUDIO_AAC => {
549                make_aac_endpoint(seid, avdtp::EndpointType::Source)
550            }
551            _ => panic!("Unsupported codec_type"),
552        };
553        Stream::build(endpoint, TestMediaTaskBuilder::new().builder())
554    }
555
556    #[fuchsia::test]
557    fn streams_basic_functionality() {
558        let mut streams = Streams::default();
559
560        streams.insert(make_stream(1, avdtp::MediaCodecType::AUDIO_SBC));
561        streams.insert(make_stream(6, avdtp::MediaCodecType::AUDIO_AAC));
562
563        let first_id = 1_u8.try_into().expect("good id");
564        let missing_id = 5_u8.try_into().expect("good id");
565
566        assert!(streams.get(&first_id).is_some());
567        assert!(streams.get(&missing_id).is_none());
568
569        assert!(streams.get_mut(&first_id).is_some());
570        assert!(streams.get_mut(&missing_id).is_none());
571
572        let expected_info = vec![
573            make_sbc_endpoint(1, avdtp::EndpointType::Source).information(),
574            make_aac_endpoint(6, avdtp::EndpointType::Source).information(),
575        ];
576
577        let infos = streams.information();
578
579        assert_eq!(expected_info.len(), infos.len());
580
581        if infos[0].id() == &first_id {
582            assert_eq!(expected_info[0], infos[0]);
583            assert_eq!(expected_info[1], infos[1]);
584        } else {
585            assert_eq!(expected_info[0], infos[1]);
586            assert_eq!(expected_info[1], infos[0]);
587        }
588    }
589
590    #[fuchsia::test]
591    fn streams_filters_compatible_codecs() {
592        let mut streams = Streams::default();
593        streams.insert(make_stream(1, avdtp::MediaCodecType::AUDIO_SBC));
594        streams.insert(make_stream(6, avdtp::MediaCodecType::AUDIO_AAC));
595
596        // Even if the other bitrate is higher, we can negotiate to the lower bitrate.
597        let config_high_bitrate_aac =
598            MediaCodecConfig::try_from(&aac_mediacodec_capability(HIGH_BITRATE)).unwrap();
599
600        let compatible: Vec<_> = streams.compatible(config_high_bitrate_aac).collect();
601        assert_eq!(compatible.len(), 1);
602        let codec_capability = compatible[0]
603            .endpoint()
604            .capabilities()
605            .iter()
606            .find(|x| x.category() == avdtp::ServiceCategory::MediaCodec)
607            .expect("should have a codec");
608        assert_eq!(
609            MediaCodecConfig::try_from(codec_capability).unwrap().codec_type(),
610            &avdtp::MediaCodecType::AUDIO_AAC
611        );
612    }
613
614    #[test_case(Transport::Socket ; "socket")]
615    #[test_case(Transport::Fidl ; "fidl")]
616    #[fuchsia::test]
617    fn rejects_unsupported_configurations(transport: Transport) {
618        // Needed to make fasync::Tasks.
619        let _exec = fasync::TestExecutor::new();
620        let mut builder = TestMediaTaskBuilder::new_reconfigurable();
621        let mut stream =
622            Stream::build(make_sbc_endpoint(1, avdtp::EndpointType::Source), builder.builder());
623
624        // the default test stream only supports 48000hz
625        let unsupported_sbc_codec_info = SbcCodecInfo::new(
626            SbcSamplingFrequency::FREQ44100HZ,
627            SbcChannelMode::JOINT_STEREO,
628            SbcBlockCount::SIXTEEN,
629            SbcSubBands::EIGHT,
630            SbcAllocation::LOUDNESS,
631            53,
632            53,
633        )
634        .expect("SBC codec info");
635
636        let unsupported_caps = vec![ServiceCapability::MediaCodec {
637            media_type: avdtp::MediaType::Audio,
638            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
639            codec_extra: unsupported_sbc_codec_info.to_bytes().to_vec(),
640        }];
641
642        let peer_id = PeerId(1);
643        let stream_id = 1.try_into().expect("StreamEndpointId");
644        let res = stream.configure(&peer_id, &stream_id, unsupported_caps.clone());
645        assert!(res.is_err());
646        assert_eq!(
647            res.err(),
648            Some((ServiceCategory::MediaCodec, ErrorCode::UnsupportedConfiguration))
649        );
650
651        assert_eq!(
652            stream.reconfigure(unsupported_caps.clone()),
653            Err((ServiceCategory::None, ErrorCode::BadState))
654        );
655
656        let supported_sbc_codec_info = SbcCodecInfo::new(
657            SbcSamplingFrequency::FREQ48000HZ,
658            SbcChannelMode::JOINT_STEREO,
659            SbcBlockCount::SIXTEEN,
660            SbcSubBands::EIGHT,
661            SbcAllocation::LOUDNESS,
662            53,
663            53,
664        )
665        .expect("SBC codec info");
666
667        let sbc_codec_cap = ServiceCapability::MediaCodec {
668            media_type: avdtp::MediaType::Audio,
669            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
670            codec_extra: supported_sbc_codec_info.to_bytes().to_vec(),
671        };
672
673        let supported_caps = vec![ServiceCapability::MediaTransport, sbc_codec_cap.clone()];
674
675        let res = stream.configure(&peer_id, &stream_id, supported_caps.clone());
676        assert!(res.is_ok());
677
678        // need to be in the open state for reconfigure
679        assert!(stream.endpoint_mut().establish().is_ok());
680        let (transport, _remote) = create_test_channels(transport);
681        match stream.endpoint_mut().receive_channel(transport) {
682            Ok(false) => {}
683            Ok(true) => panic!("Only should be expecting one channel"),
684            Err(e) => panic!("Expected channel to be accepted, got {:?}", e),
685        };
686
687        assert_eq!(
688            stream.reconfigure(unsupported_caps.clone()),
689            Err((ServiceCategory::MediaCodec, ErrorCode::UnsupportedConfiguration))
690        );
691
692        let new_codec_caps = vec![ServiceCapability::MediaCodec {
693            media_type: avdtp::MediaType::Audio,
694            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
695            codec_extra: supported_sbc_codec_info.to_bytes().to_vec(),
696        }];
697
698        assert!(stream.reconfigure(new_codec_caps.clone()).is_ok());
699
700        // Should be able to start after reconfigure, and we used the right configuration.
701        let _ = stream.start().expect("stream should start ok");
702        let task = builder.expect_task();
703        assert_eq!(task.codec_config, MediaCodecConfig::try_from(&new_codec_caps[0]).unwrap());
704    }
705
706    #[test_case(Transport::Socket ; "socket")]
707    #[test_case(Transport::Fidl ; "fidl")]
708    #[fuchsia::test]
709    fn reconfigure_runner_fails(transport: Transport) {
710        // Needed to make fasync::Tasks.
711        let _exec = fasync::TestExecutor::new();
712        let mut builder = TestMediaTaskBuilder::new();
713        let mut stream =
714            Stream::build(make_sbc_endpoint(1, avdtp::EndpointType::Source), builder.builder());
715
716        let supported_sbc_codec_info = SbcCodecInfo::new(
717            SbcSamplingFrequency::FREQ48000HZ,
718            SbcChannelMode::JOINT_STEREO,
719            SbcBlockCount::SIXTEEN,
720            SbcSubBands::EIGHT,
721            SbcAllocation::LOUDNESS,
722            53,
723            53,
724        )
725        .expect("SBC codec info");
726
727        let orig_codec_cap = ServiceCapability::MediaCodec {
728            media_type: avdtp::MediaType::Audio,
729            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
730            codec_extra: supported_sbc_codec_info.to_bytes().to_vec(),
731        };
732
733        let supported_caps = vec![ServiceCapability::MediaTransport, orig_codec_cap.clone()];
734
735        let res = stream.configure(&PeerId(1), &(1.try_into().unwrap()), supported_caps.clone());
736        assert!(res.is_ok());
737
738        // need to be in the open state for reconfigure
739        assert!(stream.endpoint_mut().establish().is_ok());
740        let (transport, _remote) = create_test_channels(transport);
741        match stream.endpoint_mut().receive_channel(transport) {
742            Ok(false) => {}
743            Ok(true) => panic!("Only should be expecting one channel"),
744            Err(e) => panic!("Expected channel to be accepted, got {:?}", e),
745        };
746
747        // Should be able to start after configure, and we used the right configuration.
748        let _ = stream.start().expect("stream should start ok");
749        let task = builder.expect_task();
750        assert_eq!(task.codec_config, MediaCodecConfig::try_from(&orig_codec_cap).unwrap());
751        stream.suspend().expect("stream should suspend ok");
752
753        // Try to reconfigure with a supported configuration, but the builder doesn't reconfigure.
754        let mono_sbc_codec_info = SbcCodecInfo::new(
755            SbcSamplingFrequency::FREQ48000HZ,
756            SbcChannelMode::MONO,
757            SbcBlockCount::SIXTEEN,
758            SbcSubBands::EIGHT,
759            SbcAllocation::LOUDNESS,
760            53,
761            53,
762        )
763        .expect("SBC codec info");
764
765        let new_codec_caps = vec![ServiceCapability::MediaCodec {
766            media_type: avdtp::MediaType::Audio,
767            codec_type: avdtp::MediaCodecType::AUDIO_SBC,
768            codec_extra: mono_sbc_codec_info.to_bytes().to_vec(),
769        }];
770
771        // Media Builder fails to reconfigure (as it's failing all reconfigures)
772        assert_eq!(
773            stream.reconfigure(new_codec_caps.clone()),
774            Err((ServiceCategory::MediaCodec, ErrorCode::UnsupportedConfiguration))
775        );
776
777        // Should be able to start after reconfigure, but it will use the old configuration.
778        let _ = stream.start().expect("stream should start ok");
779        let task = builder.expect_task();
780        assert_eq!(task.codec_config, MediaCodecConfig::try_from(&orig_codec_cap).unwrap());
781        stream.suspend().expect("stream should suspend ok")
782    }
783
784    #[test_case(Transport::Socket ; "socket")]
785    #[test_case(Transport::Fidl ; "fidl")]
786    #[fuchsia::test]
787    fn suspend_stops_media_task(transport: Transport) {
788        let mut exec = fasync::TestExecutor::new();
789
790        let mut task_builder = TestMediaTaskBuilder::new();
791        let mut stream = Stream::build(
792            make_sbc_endpoint(1, avdtp::EndpointType::Source),
793            task_builder.builder(),
794        );
795        let next_task_fut = task_builder.next_task();
796        let remote_id = 1_u8.try_into().expect("good id");
797
798        let sbc_codec_cap = sbc_mediacodec_capability();
799        let expected_codec_config =
800            MediaCodecConfig::try_from(&sbc_codec_cap).expect("codec config");
801
802        assert!(stream.configure(&PeerId(1), &remote_id, vec![]).is_err());
803        assert!(stream.configure(&PeerId(1), &remote_id, vec![sbc_codec_cap]).is_ok());
804
805        stream.endpoint_mut().establish().expect("establishment should start okay");
806        let (transport, _remote) = create_test_channels(transport);
807        let _ = stream.endpoint_mut().receive_channel(transport).expect("ready for a channel");
808
809        assert!(stream.start().is_ok());
810
811        // Task should be created here.
812        let task = {
813            let mut next_task_fut = pin!(next_task_fut);
814            match exec.run_until_stalled(&mut next_task_fut) {
815                Poll::Ready(Some(task)) => task,
816                x => panic!("Expected next task to be sent after start, got {:?}", x),
817            }
818        };
819
820        assert_eq!(task.peer_id, PeerId(1));
821        assert_eq!(task.codec_config, expected_codec_config);
822
823        assert!(task.is_started());
824        assert!(stream.suspend().is_ok());
825        assert!(!task.is_started());
826        assert!(stream.start().is_ok());
827
828        let next_task_fut = task_builder.next_task();
829        // Task should be created here.
830        let task = {
831            let mut next_task_fut = pin!(next_task_fut);
832            match exec.run_until_stalled(&mut next_task_fut) {
833                Poll::Ready(Some(task)) => task,
834                x => panic!("Expected next task to be sent after start, got {:?}", x),
835            }
836        };
837
838        assert!(task.is_started());
839    }
840
841    #[test_case(Transport::Socket ; "socket")]
842    #[test_case(Transport::Fidl ; "fidl")]
843    #[fuchsia::test]
844    fn media_task_ending_ends_future(transport: Transport) {
845        let mut exec = fasync::TestExecutor::new();
846
847        let mut task_builder = TestMediaTaskBuilder::new();
848        let mut stream = Stream::build(
849            make_sbc_endpoint(1, avdtp::EndpointType::Source),
850            task_builder.builder(),
851        );
852        let next_task_fut = task_builder.next_task();
853        let peer_id = PeerId(1);
854        let remote_id = 1_u8.try_into().expect("good id");
855
856        let sbc_codec_cap = sbc_mediacodec_capability();
857        let expected_codec_config =
858            MediaCodecConfig::try_from(&sbc_codec_cap).expect("codec config");
859
860        assert!(stream.configure(&peer_id, &remote_id, vec![]).is_err());
861        assert!(stream.configure(&peer_id, &remote_id, vec![sbc_codec_cap]).is_ok());
862
863        stream.endpoint_mut().establish().expect("establishment should start okay");
864        let (transport, _remote) = create_test_channels(transport);
865        let _ = stream.endpoint_mut().receive_channel(transport).expect("ready for a channel");
866
867        let stream_finish_fut = stream.start().expect("start to succeed with a future");
868        let mut stream_finish_fut = pin!(stream_finish_fut);
869
870        let task = {
871            let mut next_task_fut = pin!(next_task_fut);
872            match exec.run_until_stalled(&mut next_task_fut) {
873                Poll::Ready(Some(task)) => task,
874                x => panic!("Expected next task to be sent after start, got {:?}", x),
875            }
876        };
877
878        assert_eq!(task.peer_id, PeerId(1));
879        assert_eq!(task.codec_config, expected_codec_config);
880
881        // Does not need to be polled to be started.
882        assert!(task.is_started());
883
884        assert!(exec.run_until_stalled(&mut stream_finish_fut).is_pending());
885
886        task.end_prematurely(Some(Ok(())));
887        assert!(!task.is_started());
888
889        // The future should be finished, since the task ended.
890        match exec.run_until_stalled(&mut stream_finish_fut) {
891            Poll::Ready(Ok(())) => {}
892            x => panic!("Expected to get ready Ok from finish future, but got {x:?}"),
893        };
894
895        // Should still be able to suspend the stream.
896        assert!(stream.suspend().is_ok());
897
898        // And be able to restart it.
899        let result_fut = stream.start().expect("start to succeed with a future");
900
901        let next_task_fut = task_builder.next_task();
902        let mut next_task_fut = pin!(next_task_fut);
903        let task = match exec.run_until_stalled(&mut next_task_fut) {
904            Poll::Ready(Some(task)) => task,
905            x => panic!("Expected next task to be sent after restart, got {x:?}"),
906        };
907
908        assert!(task.is_started());
909
910        // Dropping the result future shouldn't stop the media task.
911        drop(result_fut);
912
913        assert!(task.is_started());
914    }
915
916    #[test_case(Transport::Socket ; "socket")]
917    #[test_case(Transport::Fidl ; "fidl")]
918    #[fuchsia::test]
919    fn set_delay_correct_results_transmits_to_task(transport: Transport) {
920        let mut _exec = fasync::TestExecutor::new();
921
922        let mut task_builder = TestMediaTaskBuilder::new_delayable();
923        let mut stream = Stream::build(
924            make_sbc_endpoint(1, avdtp::EndpointType::Source),
925            task_builder.builder(),
926        );
927        let peer_id = PeerId(1);
928        let remote_id = 1_u8.try_into().expect("good id");
929
930        let sbc_codec_cap = sbc_mediacodec_capability();
931
932        let code = stream
933            .set_delay(std::time::Duration::ZERO)
934            .expect_err("before configure, can't set a delay");
935        assert_eq!(ErrorCode::BadState, code);
936
937        assert!(stream.configure(&peer_id, &remote_id, vec![]).is_err());
938        assert!(stream.configure(&peer_id, &remote_id, vec![sbc_codec_cap]).is_ok());
939
940        let delay_set = std::time::Duration::from_nanos(0xfeed);
941
942        stream.set_delay(delay_set.clone()).expect("after configure, delay is fine");
943
944        stream.endpoint_mut().establish().expect("establishment should start okay");
945        let (transport, _remote) = create_test_channels(transport);
946        let _ = stream.endpoint_mut().receive_channel(transport).expect("ready for a channel");
947        let _stream_finish_fut = stream.start().expect("start to succeed with a future");
948
949        let media_task = task_builder.expect_task();
950        assert_eq!(delay_set, media_task.delay);
951    }
952}
bt_a2dp/stream.rs

bt_a2dp/
stream.rs
