bt_avdtp/
stream_endpoint.rs

1// Copyright 2018 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 fidl_fuchsia_bluetooth_bredr::AudioOffloadExtProxy;
6use fuchsia_async::{DurationExt, Task, TimeoutExt};
7use fuchsia_bluetooth::types::{A2dpDirection, Channel};
8use fuchsia_sync::{Mutex, RwLock};
9use futures::stream::{FusedStream, Stream};
10use futures::{FutureExt, Sink};
11use log::warn;
12use std::pin::Pin;
13use std::sync::{Arc, Weak};
14use std::task::{Context, Poll};
15use std::{fmt, io};
16use zx::{MonotonicDuration, Status};
17
18use crate::types::{
19    EndpointType, Error, ErrorCode, MediaCodecType, MediaType, Result as AvdtpResult,
20    ServiceCapability, ServiceCategory, StreamEndpointId, StreamInformation,
21};
22use crate::{Peer, SimpleResponder};
23
24pub type StreamEndpointUpdateCallback = Box<dyn Fn(&StreamEndpoint) -> () + Sync + Send>;
25
26/// The state of a StreamEndpoint.
27#[derive(PartialEq, Debug, Default, Clone, Copy)]
28pub enum StreamState {
29    #[default]
30    Idle,
31    Configured,
32    // An Open command has been accepted, but streams have not been established yet.
33    Opening,
34    Open,
35    Streaming,
36    Closing,
37    Aborting,
38}
39
40/// An AVDTP StreamEndpoint. StreamEndpoints represent a particular capability of the application
41/// to be a source of sink of media. Included here to aid negotiating the stream connection.
42/// See Section 5.3 of the AVDTP 1.3 Specification for more information about the Stream Endpoint
43/// Architecture.
44pub struct StreamEndpoint {
45    /// Local stream endpoint id.  This should be unique per AVDTP Peer.
46    id: StreamEndpointId,
47    /// The type of endpoint this is (TSEP), Source or Sink.
48    endpoint_type: EndpointType,
49    /// The media type this stream represents.
50    media_type: MediaType,
51    /// Current state the stream is in. See Section 6.5 for an overview.
52    state: Arc<Mutex<StreamState>>,
53    /// The media transport channel
54    /// This should be Some(channel) when state is Open or Streaming.
55    transport: Option<Arc<RwLock<Channel>>>,
56    /// True when the MediaStream is held.
57    /// Prevents multiple threads from owning the media stream.
58    stream_held: Arc<Mutex<bool>>,
59    /// The capabilities of this endpoint.
60    capabilities: Vec<ServiceCapability>,
61    /// The remote stream endpoint id.  None if the stream has never been configured.
62    remote_id: Option<StreamEndpointId>,
63    /// The current configuration of this endpoint.  Empty if the stream has never been configured.
64    configuration: Vec<ServiceCapability>,
65    /// Callback that is run whenever the endpoint is updated
66    update_callback: Option<StreamEndpointUpdateCallback>,
67    /// In-progress task. This is only used for the Release procedure which places the state in Closing
68    /// and must wait for the peer to close transport channels.
69    in_progress: Option<Task<()>>,
70}
71
72impl fmt::Debug for StreamEndpoint {
73    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74        f.debug_struct("StreamEndpoint")
75            .field("id", &self.id.0)
76            .field("endpoint_type", &self.endpoint_type)
77            .field("media_type", &self.media_type)
78            .field("state", &self.state)
79            .field("capabilities", &self.capabilities)
80            .field("remote_id", &self.remote_id.as_ref().map(|id| id.to_string()))
81            .field("configuration", &self.configuration)
82            .finish()
83    }
84}
85
86impl StreamEndpoint {
87    /// Make a new StreamEndpoint.
88    /// |id| must be in the valid range for a StreamEndpointId (0x01 - 0x3E).
89    /// StreamEndpoints start in the Idle state.
90    pub fn new(
91        id: u8,
92        media_type: MediaType,
93        endpoint_type: EndpointType,
94        capabilities: Vec<ServiceCapability>,
95    ) -> AvdtpResult<StreamEndpoint> {
96        let seid = StreamEndpointId::try_from(id)?;
97        Ok(StreamEndpoint {
98            id: seid,
99            capabilities,
100            media_type,
101            endpoint_type,
102            state: Default::default(),
103            transport: None,
104            stream_held: Arc::new(Mutex::new(false)),
105            remote_id: None,
106            configuration: vec![],
107            update_callback: None,
108            in_progress: None,
109        })
110    }
111
112    pub fn as_new(&self) -> Self {
113        StreamEndpoint::new(
114            self.id.0,
115            self.media_type.clone(),
116            self.endpoint_type.clone(),
117            self.capabilities.clone(),
118        )
119        .expect("as_new")
120    }
121
122    /// Set the state to the given value and run the `update_callback` afterwards
123    fn set_state(&mut self, state: StreamState) {
124        *self.state.lock() = state;
125        self.update_callback();
126    }
127
128    /// Pass update callback to StreamEndpoint that will be called anytime `StreamEndpoint` is
129    /// modified.
130    pub fn set_update_callback(&mut self, callback: Option<StreamEndpointUpdateCallback>) {
131        self.update_callback = callback;
132    }
133
134    fn update_callback(&self) {
135        if let Some(cb) = self.update_callback.as_ref() {
136            cb(self);
137        }
138    }
139
140    /// Build a new StreamEndpoint from a StreamInformation and associated Capabilities.
141    /// This makes it easy to build from AVDTP Discover and GetCapabilities procedures.
142    /// StreamEndpooints start in the Idle state.
143    pub fn from_info(
144        info: &StreamInformation,
145        capabilities: Vec<ServiceCapability>,
146    ) -> StreamEndpoint {
147        StreamEndpoint {
148            id: info.id().clone(),
149            capabilities,
150            media_type: info.media_type().clone(),
151            endpoint_type: info.endpoint_type().clone(),
152            state: Default::default(),
153            transport: None,
154            stream_held: Arc::new(Mutex::new(false)),
155            remote_id: None,
156            configuration: vec![],
157            update_callback: None,
158            in_progress: None,
159        }
160    }
161
162    /// Checks that the state is in the set of states.
163    /// If not, returns Err(ErrorCode::BadState).
164    fn state_is(&self, state: StreamState) -> Result<(), ErrorCode> {
165        (*self.state.lock() == state).then_some(()).ok_or(ErrorCode::BadState)
166    }
167
168    /// Attempt to Configure this stream using the capabilities given.
169    /// If the stream is not in an Idle state, fails with Err(SepInUse).
170    /// Used for the Stream Configuration procedure, see Section 6.9
171    pub fn configure(
172        &mut self,
173        remote_id: &StreamEndpointId,
174        capabilities: Vec<ServiceCapability>,
175    ) -> Result<(), (ServiceCategory, ErrorCode)> {
176        self.state_is(StreamState::Idle)
177            .map_err(|_| (ServiceCategory::None, ErrorCode::SepInUse))?;
178        self.remote_id = Some(remote_id.clone());
179        for cap in &capabilities {
180            if !self
181                .capabilities
182                .iter()
183                .any(|y| std::mem::discriminant(cap) == std::mem::discriminant(y))
184            {
185                return Err((cap.category(), ErrorCode::UnsupportedConfiguration));
186            }
187        }
188        self.configuration = capabilities;
189        self.set_state(StreamState::Configured);
190        Ok(())
191    }
192
193    /// Attempt to reconfigure this stream with the capabilities given.  If any capability is not
194    /// valid to set, fails with the first such category and InvalidCapabilities If the stream is
195    /// not in the Open state, fails with Err((None, BadState)) Used for the Stream Reconfiguration
196    /// procedure, see Section 6.15.
197    pub fn reconfigure(
198        &mut self,
199        mut capabilities: Vec<ServiceCapability>,
200    ) -> Result<(), (ServiceCategory, ErrorCode)> {
201        self.state_is(StreamState::Open).map_err(|e| (ServiceCategory::None, e))?;
202        // Only application capabilities are allowed to be reconfigured. See Section 8.11.1
203        if let Some(cap) = capabilities.iter().find(|x| !x.is_application()) {
204            return Err((cap.category(), ErrorCode::InvalidCapabilities));
205        }
206        // Should only replace the capabilities that have been configured. See Section 8.11.2
207        let to_replace: std::vec::Vec<_> =
208            capabilities.iter().map(|x| std::mem::discriminant(x)).collect();
209        self.configuration.retain(|x| {
210            let disc = std::mem::discriminant(x);
211            !to_replace.contains(&disc)
212        });
213        self.configuration.append(&mut capabilities);
214        self.update_callback();
215        Ok(())
216    }
217
218    /// Get the current configuration of this stream.
219    /// If the stream is not configured, returns None.
220    /// Used for the Steam Get Configuration Procedure, see Section 6.10
221    pub fn get_configuration(&self) -> Option<&Vec<ServiceCapability>> {
222        if self.configuration.is_empty() {
223            return None;
224        }
225        Some(&self.configuration)
226    }
227
228    // 100 milliseconds chosen based on end of range testing, to allow for recovery after normal
229    // packet delivery continues.
230    const SRC_FLUSH_TIMEOUT: MonotonicDuration = MonotonicDuration::from_millis(100);
231
232    /// When a L2CAP channel is received after an Open command is accepted, it should be
233    /// delivered via receive_channel.
234    /// Returns true if this Endpoint expects more channels to be established before
235    /// streaming is started.
236    /// Returns Err(InvalidState) if this Endpoint is not expecting a channel to be established,
237    /// closing |c|.
238    pub fn receive_channel(&mut self, c: Channel) -> AvdtpResult<bool> {
239        if self.state_is(StreamState::Opening).is_err() || self.transport.is_some() {
240            return Err(Error::InvalidState);
241        }
242        self.transport = Some(Arc::new(RwLock::new(c)));
243        self.try_flush_timeout(Self::SRC_FLUSH_TIMEOUT);
244        self.stream_held = Arc::new(Mutex::new(false));
245        // TODO(jamuraa, https://fxbug.dev/42051664, https://fxbug.dev/42051776): Reporting and Recovery channels
246        self.set_state(StreamState::Open);
247        Ok(false)
248    }
249
250    /// Begin opening this stream.  The stream must be in a Configured state.
251    /// See Stream Establishment, Section 6.11
252    pub fn establish(&mut self) -> Result<(), ErrorCode> {
253        if self.state_is(StreamState::Configured).is_err() || self.transport.is_some() {
254            return Err(ErrorCode::BadState);
255        }
256        self.set_state(StreamState::Opening);
257        Ok(())
258    }
259
260    /// Attempts to set audio direction priority of the MediaTransport channel based on
261    /// whether the stream is a source or sink endpoint if `active` is true.  If `active` is
262    /// false, set the priority to Normal instead.  Does nothing on failure.
263    pub fn try_priority(&self, active: bool) {
264        let priority = match (active, &self.endpoint_type) {
265            (false, _) => A2dpDirection::Normal,
266            (true, EndpointType::Source) => A2dpDirection::Source,
267            (true, EndpointType::Sink) => A2dpDirection::Sink,
268        };
269        let fut = match self.transport.as_ref().unwrap().try_read() {
270            None => return,
271            Some(channel) => channel.set_audio_priority(priority).map(|_| ()),
272        };
273        // TODO(https://fxbug.dev/331621666): We should avoid detaching this.
274        Task::spawn(fut).detach();
275    }
276
277    /// Attempts to set the flush timeout for the MediaTransport channel, for source endpoints.
278    pub fn try_flush_timeout(&self, timeout: MonotonicDuration) {
279        if self.endpoint_type != EndpointType::Source {
280            return;
281        }
282        let fut = match self.transport.as_ref().unwrap().try_write() {
283            None => return,
284            Some(channel) => channel.set_flush_timeout(Some(timeout)).map(|_| ()),
285        };
286        // TODO(https://fxbug.dev/331621666): We should avoid detaching this.
287        Task::spawn(fut).detach();
288    }
289
290    /// Close this stream.  This procedure will wait until media channels are closed before
291    /// transitioning to Idle.  If the channels are not closed in 3 seconds, we initiate an abort
292    /// procedure with the remote |peer| to force a transition to Idle.
293    pub fn release(&mut self, responder: SimpleResponder, peer: &Peer) -> AvdtpResult<()> {
294        {
295            let lock = self.state.lock();
296            if *lock != StreamState::Open && *lock != StreamState::Streaming {
297                return responder.reject(ErrorCode::BadState);
298            }
299        }
300        self.set_state(StreamState::Closing);
301        responder.send()?;
302        let release_wait_fut = {
303            // Take our transport and remote id - after this procedure it will be closed.
304            // These must be Some(_) because we are in Open / Streaming state.
305            let seid = self.remote_id.take().unwrap();
306            let transport = self.transport.take().unwrap();
307            let peer = peer.clone();
308            let state = self.state.clone();
309            async move {
310                let closed_fut = {
311                    let Some(channel) = transport.try_read() else {
312                        warn!("unable to lock transport channel, dropping and assuming closed");
313                        *state.lock() = StreamState::Idle;
314                        return;
315                    };
316                    channel.closed()
317                };
318
319                let closed_fut = closed_fut
320                    .on_timeout(MonotonicDuration::from_seconds(3).after_now(), || {
321                        Err(Status::TIMED_OUT)
322                    });
323
324                if let Err(Status::TIMED_OUT) = closed_fut.await {
325                    let _ = peer.abort(&seid).await;
326                    *state.lock() = StreamState::Aborting;
327                }
328                // Dropping the Arc<RwLock<Channel>> closes our end of the transport.
329                drop(transport);
330                *state.lock() = StreamState::Idle;
331            }
332        };
333        self.in_progress = Some(Task::local(release_wait_fut));
334        // Closing will return this endpoint to the Idle state, one way or another with no
335        // configuration
336        self.configuration.clear();
337        self.update_callback();
338        Ok(())
339    }
340
341    /// Returns the current state of this endpoint.
342    pub fn state(&self) -> StreamState {
343        *self.state.lock()
344    }
345
346    /// Start this stream.  This can be done only from the Open State.
347    /// Used for the Stream Start procedure, See Section 6.12
348    pub fn start(&mut self) -> Result<(), ErrorCode> {
349        self.state_is(StreamState::Open)?;
350        self.try_priority(true);
351        self.set_state(StreamState::Streaming);
352        Ok(())
353    }
354
355    /// Suspend this stream.  This can be done only from the Streaming state.
356    /// Used for the Stream Suspend procedure, See Section 6.14
357    pub fn suspend(&mut self) -> Result<(), ErrorCode> {
358        self.state_is(StreamState::Streaming)?;
359        self.set_state(StreamState::Open);
360        self.try_priority(false);
361        Ok(())
362    }
363
364    /// Abort this stream.  This can be done from any state, and will always return the state
365    /// to Idle.  We are initiating this procedure so will wait for a response and all our
366    /// channels will be closed.
367    pub async fn initiate_abort<'a>(&'a mut self, peer: &'a Peer) {
368        if let Some(seid) = self.remote_id.take() {
369            let _ = peer.abort(&seid).await;
370            self.set_state(StreamState::Aborting);
371        }
372        self.abort()
373    }
374
375    /// Abort this stream.  This can be done from any state, and will always return the state
376    /// to Idle.  We are receiving this abort from the peer, and all our channels will close.
377    pub fn abort(&mut self) {
378        self.set_state(StreamState::Aborting);
379        self.configuration.clear();
380        self.remote_id = None;
381        self.transport = None;
382        self.set_state(StreamState::Idle);
383    }
384
385    /// Capabilities of this StreamEndpoint.
386    /// Provides support for the Get Capabilities and Get All Capabilities signaling procedures.
387    /// See Sections 6.7 and 6.8
388    pub fn capabilities(&self) -> &Vec<ServiceCapability> {
389        &self.capabilities
390    }
391
392    /// Returns the CodecType of this StreamEndpoint.
393    /// Returns None if there is no MediaCodec capability in the endpoint.
394    /// Note: a MediaCodec capability is required by all endpoints by the spec.
395    pub fn codec_type(&self) -> Option<&MediaCodecType> {
396        self.capabilities.iter().find_map(|cap| match cap {
397            ServiceCapability::MediaCodec { codec_type, .. } => Some(codec_type),
398            _ => None,
399        })
400    }
401
402    /// Returns the local StreamEndpointId for this endpoint.
403    pub fn local_id(&self) -> &StreamEndpointId {
404        &self.id
405    }
406
407    /// Returns the remote StreamEndpointId for this endpoint, if it's configured.
408    pub fn remote_id(&self) -> Option<&StreamEndpointId> {
409        self.remote_id.as_ref()
410    }
411
412    /// Returns the EndpointType of this endpoint
413    pub fn endpoint_type(&self) -> &EndpointType {
414        &self.endpoint_type
415    }
416
417    /// Make a StreamInformation which represents the current state of this stream.
418    pub fn information(&self) -> StreamInformation {
419        let in_use = self.state_is(StreamState::Idle).is_err();
420        StreamInformation::new(
421            self.id.clone(),
422            in_use,
423            self.media_type.clone(),
424            self.endpoint_type.clone(),
425        )
426    }
427
428    /// Take the media transport channel, which transmits (or receives) any media for this
429    /// StreamEndpoint.  Returns None if the channel is held already, or if the channel has not
430    /// been opened.
431    pub fn take_transport(&mut self) -> Option<MediaStream> {
432        let mut stream_held = self.stream_held.lock();
433        if *stream_held || self.transport.is_none() {
434            return None;
435        }
436
437        *stream_held = true;
438
439        Some(MediaStream::new(
440            self.stream_held.clone(),
441            Arc::downgrade(self.transport.as_ref().unwrap()),
442        ))
443    }
444
445    /// Get an AudioOffloadExtProxy if it exists in the transport channel
446    pub fn audio_offload(&self) -> Option<AudioOffloadExtProxy> {
447        self.transport.as_ref().and_then(|c| c.read().audio_offload())
448    }
449}
450
451/// Represents a media transport stream.
452/// If a sink, produces the bytes that have been delivered from the peer.
453/// If a source, can send bytes using `send`
454pub struct MediaStream {
455    in_use: Arc<Mutex<bool>>,
456    channel: Weak<RwLock<Channel>>,
457    terminated: bool,
458}
459
460impl MediaStream {
461    pub fn new(in_use: Arc<Mutex<bool>>, channel: Weak<RwLock<Channel>>) -> Self {
462        Self { in_use, channel, terminated: false }
463    }
464
465    fn try_upgrade(&self) -> Result<Arc<RwLock<Channel>>, io::Error> {
466        self.channel
467            .upgrade()
468            .ok_or_else(|| io::Error::new(io::ErrorKind::ConnectionAborted, "lost connection"))
469    }
470
471    pub fn max_tx_size(&self) -> Result<usize, io::Error> {
472        match self.try_upgrade()?.try_read() {
473            None => return Err(io::Error::new(io::ErrorKind::WouldBlock, "couldn't lock")),
474            Some(lock) => Ok(lock.max_tx_size()),
475        }
476    }
477}
478
479impl Drop for MediaStream {
480    fn drop(&mut self) {
481        let mut l = self.in_use.lock();
482        *l = false;
483    }
484}
485
486impl Stream for MediaStream {
487    type Item = AvdtpResult<Vec<u8>>;
488
489    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
490        let Ok(arc_chan) = self.try_upgrade() else {
491            self.terminated = true;
492            return Poll::Ready(None);
493        };
494        let Some(lock) = arc_chan.try_write() else {
495            self.terminated = true;
496            return Poll::Ready(None);
497        };
498        let mut pin_chan = Pin::new(lock);
499        match pin_chan.as_mut().poll_next(cx) {
500            Poll::Ready(Some(Ok(res))) => Poll::Ready(Some(Ok(res))),
501            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(Error::PeerRead(e)))),
502            Poll::Ready(None) => {
503                self.terminated = true;
504                Poll::Ready(None)
505            }
506            Poll::Pending => Poll::Pending,
507        }
508    }
509}
510
511impl FusedStream for MediaStream {
512    fn is_terminated(&self) -> bool {
513        self.terminated
514    }
515}
516
517impl Sink<Vec<u8>> for MediaStream {
518    type Error = io::Error;
519
520    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
521        let arc_chan = self.try_upgrade()?;
522        let mut lock = arc_chan
523            .try_write()
524            .ok_or_else(|| io::Error::new(io::ErrorKind::WouldBlock, "couldn't lock"))?;
525        Pin::new(&mut *lock).poll_ready(cx).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
526    }
527
528    fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
529        let arc_chan = self.try_upgrade()?;
530        let mut lock = arc_chan
531            .try_write()
532            .ok_or_else(|| io::Error::new(io::ErrorKind::WouldBlock, "couldn't lock"))?;
533        Pin::new(&mut *lock).start_send(item).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
534    }
535
536    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
537        let arc_chan = self.try_upgrade()?;
538        let mut lock = arc_chan
539            .try_write()
540            .ok_or_else(|| io::Error::new(io::ErrorKind::WouldBlock, "couldn't lock"))?;
541        Pin::new(&mut *lock).poll_flush(cx).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
542    }
543
544    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
545        let arc_chan = self.try_upgrade()?;
546        let mut lock = arc_chan
547            .try_write()
548            .ok_or_else(|| io::Error::new(io::ErrorKind::WouldBlock, "couldn't lock"))?;
549        Pin::new(&mut *lock).poll_close(cx).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
550    }
551}
552
553#[cfg(test)]
554mod tests {
555    use super::*;
556    use crate::Request;
557    use crate::tests::{expect_remote_recv, setup_peer};
558    use bt_channel_test_support::{Transport, create_test_channels};
559    use test_case::test_case;
560
561    use assert_matches::assert_matches;
562    use async_utils::PollExt;
563    use fidl::endpoints::create_request_stream;
564    use fidl_fuchsia_bluetooth as fidl_bt;
565    use fidl_fuchsia_bluetooth_bredr as bredr;
566    use fuchsia_async as fasync;
567    use futures::SinkExt;
568    use futures::stream::StreamExt;
569
570    const REMOTE_ID_VAL: u8 = 1;
571    const REMOTE_ID: StreamEndpointId = StreamEndpointId(REMOTE_ID_VAL);
572
573    #[test]
574    fn make() {
575        let s = StreamEndpoint::new(
576            REMOTE_ID_VAL,
577            MediaType::Audio,
578            EndpointType::Sink,
579            vec![ServiceCapability::MediaTransport],
580        );
581        assert!(s.is_ok());
582        let s = s.unwrap();
583        assert_eq!(&StreamEndpointId(1), s.local_id());
584
585        let info = s.information();
586        assert!(!info.in_use());
587
588        let no = StreamEndpoint::new(
589            0,
590            MediaType::Audio,
591            EndpointType::Sink,
592            vec![ServiceCapability::MediaTransport],
593        );
594        assert!(no.is_err());
595    }
596
597    fn establish_stream(s: &mut StreamEndpoint, transport: Transport) -> Channel {
598        assert_matches!(s.establish(), Ok(()));
599        let (chan, remote) = create_test_channels(transport);
600        assert_matches!(s.receive_channel(chan), Ok(false));
601        remote
602    }
603
604    #[test]
605    fn from_info() {
606        let seid = StreamEndpointId::try_from(5).unwrap();
607        let info =
608            StreamInformation::new(seid.clone(), false, MediaType::Audio, EndpointType::Sink);
609        let capabilities = vec![ServiceCapability::MediaTransport];
610
611        let endpoint = StreamEndpoint::from_info(&info, capabilities);
612
613        assert_eq!(&seid, endpoint.local_id());
614        assert_eq!(&false, endpoint.information().in_use());
615        assert_eq!(1, endpoint.capabilities().len());
616    }
617
618    #[test]
619    fn codec_type() {
620        let s = StreamEndpoint::new(
621            REMOTE_ID_VAL,
622            MediaType::Audio,
623            EndpointType::Sink,
624            vec![
625                ServiceCapability::MediaTransport,
626                ServiceCapability::MediaCodec {
627                    media_type: MediaType::Audio,
628                    codec_type: MediaCodecType::new(0x40),
629                    codec_extra: vec![0xDE, 0xAD, 0xBE, 0xEF], // Meaningless test data.
630                },
631            ],
632        )
633        .unwrap();
634
635        assert_eq!(Some(&MediaCodecType::new(0x40)), s.codec_type());
636
637        let s = StreamEndpoint::new(
638            REMOTE_ID_VAL,
639            MediaType::Audio,
640            EndpointType::Sink,
641            vec![ServiceCapability::MediaTransport],
642        )
643        .unwrap();
644
645        assert_eq!(None, s.codec_type());
646    }
647
648    fn test_endpoint(r#type: EndpointType) -> StreamEndpoint {
649        StreamEndpoint::new(
650            REMOTE_ID_VAL,
651            MediaType::Audio,
652            r#type,
653            vec![
654                ServiceCapability::MediaTransport,
655                ServiceCapability::MediaCodec {
656                    media_type: MediaType::Audio,
657                    codec_type: MediaCodecType::new(0x40),
658                    codec_extra: vec![0xDE, 0xAD, 0xBE, 0xEF], // Meaningless test data.
659                },
660            ],
661        )
662        .unwrap()
663    }
664
665    #[test_case(Transport::Socket ; "socket")]
666    #[test_case(Transport::Fidl ; "fidl")]
667    #[fuchsia::test]
668    fn stream_configure_reconfigure(transport: Transport) {
669        let _exec = fasync::TestExecutor::new();
670        let mut s = test_endpoint(EndpointType::Sink);
671
672        // Can't configure items that aren't in range.
673        assert_matches!(
674            s.configure(&REMOTE_ID, vec![ServiceCapability::Reporting]),
675            Err((ServiceCategory::Reporting, ErrorCode::UnsupportedConfiguration))
676        );
677
678        assert_matches!(
679            s.configure(
680                &REMOTE_ID,
681                vec![
682                    ServiceCapability::MediaTransport,
683                    ServiceCapability::MediaCodec {
684                        media_type: MediaType::Audio,
685                        codec_type: MediaCodecType::new(0x40),
686                        // Change the codec_extra which is typical, ex. SBC (A2DP Spec 4.3.2.6)
687                        codec_extra: vec![0x0C, 0x0D, 0x02, 0x51],
688                    }
689                ]
690            ),
691            Ok(())
692        );
693
694        // Configuring not allowed when not IDLE
695        assert_matches!(
696            s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]),
697            Err((_, ErrorCode::SepInUse))
698        );
699
700        // Can't configure while open
701        let _channel = establish_stream(&mut s, transport);
702
703        assert_matches!(
704            s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]),
705            Err((_, ErrorCode::SepInUse))
706        );
707
708        let reconfiguration = vec![ServiceCapability::MediaCodec {
709            media_type: MediaType::Audio,
710            codec_type: MediaCodecType::new(0x40),
711            // Reconfigure to yet another different codec_extra value.
712            codec_extra: vec![0x0C, 0x0D, 0x0E, 0x0F],
713        }];
714
715        // The new configuration should match the previous one, but with the reconfigured
716        // capabilities updated.
717        let new_configuration = vec![ServiceCapability::MediaTransport, reconfiguration[0].clone()];
718
719        // Reconfiguring while open is fine though.
720        assert_matches!(s.reconfigure(reconfiguration.clone()), Ok(()));
721
722        assert_eq!(Some(&new_configuration), s.get_configuration());
723
724        // Can't reconfigure non-application types
725        assert_matches!(
726            s.reconfigure(vec![ServiceCapability::MediaTransport]),
727            Err((ServiceCategory::MediaTransport, ErrorCode::InvalidCapabilities))
728        );
729
730        // Can't configure or reconfigure while streaming
731        assert_matches!(s.start(), Ok(()));
732
733        assert_matches!(
734            s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]),
735            Err((_, ErrorCode::SepInUse))
736        );
737
738        assert_matches!(s.reconfigure(reconfiguration.clone()), Err((_, ErrorCode::BadState)));
739
740        assert_matches!(s.suspend(), Ok(()));
741
742        // Reconfigure should be fine again in open state.
743        assert_matches!(s.reconfigure(reconfiguration.clone()), Ok(()));
744
745        // Configure is still not allowed.
746        assert_matches!(
747            s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]),
748            Err((_, ErrorCode::SepInUse))
749        );
750    }
751
752    #[test_case(Transport::Socket ; "socket")]
753    #[test_case(Transport::Fidl ; "fidl")]
754    #[fuchsia::test]
755    fn stream_establishment(transport: Transport) {
756        let mut exec = fasync::TestExecutor::new();
757        let mut s = test_endpoint(EndpointType::Sink);
758
759        let (transport_chan, mut remote) = create_test_channels(transport);
760
761        // Can't establish before configuring
762        assert_matches!(s.establish(), Err(ErrorCode::BadState));
763
764        // Trying to receive a channel in the wrong state closes the channel
765        assert_matches!(s.receive_channel(transport_chan), Err(Error::InvalidState));
766
767        let mut read_fut = remote.next();
768        let res = exec.run_until_stalled(&mut read_fut).expect("should be ready");
769        // When the peer is closed, None is returned.
770        assert_matches!(res, None);
771
772        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
773
774        assert_matches!(s.establish(), Ok(()));
775
776        // And we should be able to give a channel now.
777        let (transport_chan, _remote) = create_test_channels(transport);
778        assert_matches!(s.receive_channel(transport_chan), Ok(false));
779    }
780
781    fn setup_peer_for_release(
782        exec: &mut fasync::TestExecutor,
783        transport: Transport,
784    ) -> (Peer, Channel, SimpleResponder) {
785        let (peer, mut signaling) = setup_peer(transport);
786        // Send a close from the other side to produce an event we can respond to.
787        exec.run_until_stalled(&mut signaling.send(vec![0x40, 0x08, 0x04]))
788            .expect("signaling write")
789            .expect("write successful");
790        let mut req_stream = peer.take_request_stream();
791        let mut req_fut = req_stream.next();
792        let complete = exec.run_until_stalled(&mut req_fut);
793        let responder = match complete {
794            Poll::Ready(Some(Ok(Request::Close { responder, .. }))) => responder,
795            _ => panic!("Expected a close request"),
796        };
797        (peer, signaling, responder)
798    }
799
800    #[test_case(Transport::Socket ; "socket")]
801    #[test_case(Transport::Fidl ; "fidl")]
802    #[fuchsia::test]
803    fn stream_release_without_abort(transport: Transport) {
804        let mut exec = fasync::TestExecutor::new();
805        let mut s = test_endpoint(EndpointType::Sink);
806
807        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
808
809        let remote_transport = establish_stream(&mut s, transport);
810
811        let (peer, mut signaling, responder) = setup_peer_for_release(&mut exec, transport);
812
813        // We expect release to succeed in this state.
814        s.release(responder, &peer).unwrap();
815        // Expect a "yes" response.
816        expect_remote_recv(&mut exec, &[0x42, 0x08], &mut signaling);
817
818        // Close the transport channel by dropping it.
819        drop(remote_transport);
820
821        // After the transport is closed we should transition to Idle.
822        let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
823        assert_eq!(s.state(), StreamState::Idle);
824    }
825
826    #[test_case(Transport::Socket ; "socket")]
827    #[test_case(Transport::Fidl ; "fidl")]
828    #[fuchsia::test]
829    fn test_mediastream(transport: Transport) {
830        let mut exec = fasync::TestExecutor::new();
831        let mut s = test_endpoint(EndpointType::Sink);
832
833        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
834
835        // Before the stream is opened, we shouldn't be able to take the transport.
836        assert!(s.take_transport().is_none());
837
838        let mut remote_transport = establish_stream(&mut s, transport);
839
840        // Should be able to get the transport from the stream now.
841        let temp_stream = s.take_transport();
842        assert!(temp_stream.is_some());
843
844        // But only once
845        assert!(s.take_transport().is_none());
846
847        // Until you drop the stream
848        drop(temp_stream);
849
850        let media_stream = s.take_transport();
851        assert!(media_stream.is_some());
852        let mut media_stream = media_stream.unwrap();
853
854        // Max TX size is taken from the underlying channel.
855        assert_matches!(media_stream.max_tx_size(), Ok(Channel::DEFAULT_MAX_TX));
856
857        // Writing to the media stream should send it through the transport channel.
858        let hearts = vec![0xF0, 0x9F, 0x92, 0x96, 0xF0, 0x9F, 0x92, 0x96];
859        let mut write_fut = media_stream.send(hearts.clone());
860
861        assert_matches!(exec.run_until_stalled(&mut write_fut), Poll::Ready(Ok(())));
862
863        expect_remote_recv(&mut exec, &hearts, &mut remote_transport);
864
865        // Closing the media stream should close the channel.
866        let mut close_fut = media_stream.close();
867        assert_matches!(exec.run_until_stalled(&mut close_fut), Poll::Ready(Ok(())));
868        // Note: there's no effect on the other end of the channel when a close occurs,
869        // until the channel is dropped.
870
871        drop(s);
872
873        // Reading from the remote end should return None.
874        let mut read_fut = remote_transport.next();
875        let res = exec.run_until_stalled(&mut read_fut).expect("should be ready");
876        // When the peer is closed, None is returned.
877        assert_matches!(res, None);
878
879        // After the stream is gone, any write should return an Err
880        let mut write_fut = media_stream.send(vec![0xDE, 0xAD]);
881        assert_matches!(exec.run_until_stalled(&mut write_fut), Poll::Ready(Err(_)));
882
883        // After the stream is gone, the stream should be fused done.
884        let mut next_fut = media_stream.next();
885        assert_matches!(exec.run_until_stalled(&mut next_fut), Poll::Ready(None));
886
887        assert!(media_stream.is_terminated(), "should be terminated");
888
889        // And the Max TX should be an error.
890        assert_matches!(media_stream.max_tx_size(), Err(_));
891    }
892
893    #[test_case(Transport::Socket ; "socket")]
894    #[test_case(Transport::Fidl ; "fidl")]
895    #[fuchsia::test]
896    fn stream_release_with_abort(transport: Transport) {
897        let mut exec = fasync::TestExecutor::new();
898        let mut s = test_endpoint(EndpointType::Sink);
899
900        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
901        let remote_transport = establish_stream(&mut s, transport);
902        let (peer, mut signaling, responder) = setup_peer_for_release(&mut exec, transport);
903
904        // We expect release to succeed in this state, then start the task to wait for the close.
905        s.release(responder, &peer).unwrap();
906        // Expect a "yes" response.
907        expect_remote_recv(&mut exec, &[0x42, 0x08], &mut signaling);
908
909        // Should get an abort
910        let next = std::pin::pin!(signaling.next());
911        let received =
912            exec.run_singlethreaded(next).expect("channel not closed").expect("successful read");
913        assert_eq!(0x0A, received[1]);
914        let txlabel = received[0] & 0xF0;
915        // Send a response
916        exec.run_until_stalled(&mut signaling.send(vec![txlabel | 0x02, 0x0A]))
917            .expect("signaling write")
918            .expect("write successful");
919
920        let _ = exec.run_singlethreaded(&mut std::pin::pin!(remote_transport.closed()));
921
922        // We will then end up in Idle.
923        while s.state() != StreamState::Idle {
924            let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
925        }
926    }
927
928    fn create_channel_for_start_test(
929        transport: Transport,
930    ) -> (Channel, Channel, Option<bredr::AudioDirectionExtRequestStream>) {
931        match transport {
932            Transport::Socket => {
933                let (remote, local) = zx::Socket::create_datagram();
934                let (client_end, direction_request_stream) =
935                    create_request_stream::<bredr::AudioDirectionExtMarker>();
936                let ext = bredr::Channel {
937                    socket: Some(local),
938                    channel_mode: Some(fidl_bt::ChannelMode::Basic),
939                    max_tx_sdu_size: Some(1004),
940                    ext_direction: Some(client_end),
941                    ..Default::default()
942                };
943                let channel = Channel::try_from(ext).unwrap();
944                let remote_chan = Channel::from_socket_infallible(remote, Channel::DEFAULT_MAX_TX);
945                (channel, remote_chan, Some(direction_request_stream))
946            }
947            Transport::Fidl => {
948                let (client, server) = create_test_channels(Transport::Fidl);
949                (client, server, None)
950            }
951        }
952    }
953
954    #[test_case(Transport::Socket ; "socket")]
955    #[test_case(Transport::Fidl ; "fidl")]
956    #[fuchsia::test]
957    fn start_and_suspend(transport: Transport) {
958        let mut exec = fasync::TestExecutor::new();
959        let mut s = test_endpoint(EndpointType::Sink);
960
961        // Can't start or suspend until configured and open.
962        assert_matches!(s.start(), Err(ErrorCode::BadState));
963        assert_matches!(s.suspend(), Err(ErrorCode::BadState));
964
965        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
966
967        assert_matches!(s.start(), Err(ErrorCode::BadState));
968        assert_matches!(s.suspend(), Err(ErrorCode::BadState));
969
970        assert_matches!(s.establish(), Ok(()));
971
972        assert_matches!(s.start(), Err(ErrorCode::BadState));
973        assert_matches!(s.suspend(), Err(ErrorCode::BadState));
974
975        let (transport_chan, remote, mut direction_request_stream) =
976            create_channel_for_start_test(transport);
977        assert_matches!(s.receive_channel(transport_chan), Ok(false));
978
979        // Should be able to start but not suspend now.
980        assert_matches!(s.suspend(), Err(ErrorCode::BadState));
981        assert_matches!(s.start(), Ok(()));
982
983        if let Some(ref mut stream) = direction_request_stream {
984            match exec.run_until_stalled(&mut stream.next()) {
985                Poll::Ready(Some(Ok(bredr::AudioDirectionExtRequest::SetPriority {
986                    priority,
987                    responder,
988                }))) => {
989                    assert_eq!(bredr::A2dpDirectionPriority::Sink, priority);
990                    responder.send(Ok(())).expect("response to send cleanly");
991                }
992                x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
993            }
994        }
995
996        // Are started, so we should be able to suspend but not start again here.
997        assert_matches!(s.start(), Err(ErrorCode::BadState));
998        assert_matches!(s.suspend(), Ok(()));
999
1000        if let Some(ref mut stream) = direction_request_stream {
1001            match exec.run_until_stalled(&mut stream.next()) {
1002                Poll::Ready(Some(Ok(bredr::AudioDirectionExtRequest::SetPriority {
1003                    priority,
1004                    responder,
1005                }))) => {
1006                    assert_eq!(bredr::A2dpDirectionPriority::Normal, priority);
1007                    responder.send(Ok(())).expect("response to send cleanly");
1008                }
1009                x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
1010            }
1011        }
1012
1013        // Now we're suspended, so we can start it again.
1014        assert_matches!(s.start(), Ok(()));
1015        assert_matches!(s.suspend(), Ok(()));
1016
1017        // After we close, we are back at idle and can't start / stop
1018        let (peer, mut signaling, responder) = setup_peer_for_release(&mut exec, transport);
1019
1020        {
1021            s.release(responder, &peer).unwrap();
1022            // Expect a "yes" response.
1023            expect_remote_recv(&mut exec, &[0x42, 0x08], &mut signaling);
1024            // Close the transport channel by dropping it.
1025            drop(remote);
1026            while s.state() != StreamState::Idle {
1027                let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
1028            }
1029        }
1030
1031        // Shouldn't be able to start or suspend again.
1032        assert_matches!(s.start(), Err(ErrorCode::BadState));
1033        assert_matches!(s.suspend(), Err(ErrorCode::BadState));
1034    }
1035
1036    fn receive_l2cap_params_channel(
1037        s: &mut StreamEndpoint,
1038        transport: Transport,
1039    ) -> (Channel, Option<bredr::L2capParametersExtRequestStream>) {
1040        assert_matches!(s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport]), Ok(()));
1041        assert_matches!(s.establish(), Ok(()));
1042
1043        match transport {
1044            Transport::Socket => {
1045                let (remote, local) = zx::Socket::create_datagram();
1046                let (client_end, l2cap_params_requests) =
1047                    create_request_stream::<bredr::L2capParametersExtMarker>();
1048                let ext = bredr::Channel {
1049                    socket: Some(local),
1050                    channel_mode: Some(fidl_bt::ChannelMode::Basic),
1051                    max_tx_sdu_size: Some(1004),
1052                    ext_l2cap: Some(client_end),
1053                    ..Default::default()
1054                };
1055                let transport_chan = Channel::try_from(ext).unwrap();
1056                assert_matches!(s.receive_channel(transport_chan), Ok(false));
1057                let remote_chan = Channel::from_socket_infallible(remote, Channel::DEFAULT_MAX_TX);
1058                (remote_chan, Some(l2cap_params_requests))
1059            }
1060            Transport::Fidl => {
1061                let (client, server) = create_test_channels(Transport::Fidl);
1062                assert_matches!(s.receive_channel(client), Ok(false));
1063                (server, None)
1064            }
1065        }
1066    }
1067
1068    #[test_case(Transport::Socket ; "socket")]
1069    #[test_case(Transport::Fidl ; "fidl")]
1070    #[fuchsia::test]
1071    fn sets_flush_timeout_for_source_transports(transport: Transport) {
1072        let mut exec = fasync::TestExecutor::new();
1073        let mut s = test_endpoint(EndpointType::Source);
1074        let (_remote, mut l2cap_params_requests) = receive_l2cap_params_channel(&mut s, transport);
1075
1076        if let Some(ref mut stream) = l2cap_params_requests {
1077            match exec.run_until_stalled(&mut stream.next()) {
1078                Poll::Ready(Some(Ok(bredr::L2capParametersExtRequest::RequestParameters {
1079                    request,
1080                    responder,
1081                }))) => {
1082                    assert_eq!(
1083                        Some(StreamEndpoint::SRC_FLUSH_TIMEOUT.into_nanos()),
1084                        request.flush_timeout
1085                    );
1086                    responder.send(&request).expect("response to send cleanly");
1087                }
1088                x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
1089            };
1090        }
1091    }
1092
1093    #[test_case(Transport::Socket ; "socket")]
1094    #[test_case(Transport::Fidl ; "fidl")]
1095    #[fuchsia::test]
1096    fn no_flush_timeout_for_sink_transports(transport: Transport) {
1097        let mut exec = fasync::TestExecutor::new();
1098        let mut s = test_endpoint(EndpointType::Sink);
1099        let (_remote, mut l2cap_params_requests) = receive_l2cap_params_channel(&mut s, transport);
1100
1101        if let Some(ref mut stream) = l2cap_params_requests {
1102            // Should NOT request to set the flush timeout.
1103            match exec.run_until_stalled(&mut stream.next()) {
1104                Poll::Pending => {}
1105                x => panic!("Expected no request to set flush timeout, got {:?}", x),
1106            };
1107        }
1108    }
1109
1110    #[test]
1111    fn get_configuration() {
1112        let mut s = test_endpoint(EndpointType::Sink);
1113
1114        // Can't get configuration if we aren't configured.
1115        assert!(s.get_configuration().is_none());
1116
1117        let config = vec![
1118            ServiceCapability::MediaTransport,
1119            ServiceCapability::MediaCodec {
1120                media_type: MediaType::Audio,
1121                codec_type: MediaCodecType::new(0),
1122                // Change the codec_extra which is typical, ex. SBC (A2DP Spec 4.3.2.6)
1123                codec_extra: vec![0x60, 0x0D, 0x02, 0x55],
1124            },
1125        ];
1126
1127        assert_matches!(s.configure(&REMOTE_ID, config.clone()), Ok(()));
1128
1129        match s.get_configuration() {
1130            Some(c) => assert_eq!(&config, c),
1131            x => panic!("Expected Ok from get_configuration but got {:?}", x),
1132        };
1133
1134        // Abort this stream, putting it back to the idle state.
1135        s.abort();
1136
1137        assert!(s.get_configuration().is_none());
1138    }
1139
1140    use std::sync::atomic::{AtomicUsize, Ordering};
1141
1142    /// Create a callback that tracks how many times it has been called
1143    fn call_count_callback() -> (Option<StreamEndpointUpdateCallback>, Arc<AtomicUsize>) {
1144        let call_count = Arc::new(AtomicUsize::new(0));
1145        let call_count_reader = call_count.clone();
1146        let count_cb: StreamEndpointUpdateCallback = Box::new(move |_stream: &StreamEndpoint| {
1147            let _ = call_count.fetch_add(1, Ordering::SeqCst);
1148        });
1149        (Some(count_cb), call_count_reader)
1150    }
1151
1152    /// Test that the update callback is run at least once for all methods that mutate the state of
1153    /// the StreamEndpoint. This is done through an atomic counter in the callback that increments
1154    /// when the callback is run.
1155    ///
1156    /// Note that the _results_ of calling these mutating methods on the state of StreamEndpoint are
1157    /// not validated here. They are validated in other tests.
1158    #[test_case(Transport::Socket ; "socket")]
1159    #[test_case(Transport::Fidl ; "fidl")]
1160    #[fuchsia::test]
1161    fn update_callback(transport: Transport) {
1162        // Need an executor to make a socket
1163        let _exec = fasync::TestExecutor::new();
1164        let mut s = test_endpoint(EndpointType::Sink);
1165        let (cb, call_count) = call_count_callback();
1166        s.set_update_callback(cb);
1167
1168        s.configure(&REMOTE_ID, vec![ServiceCapability::MediaTransport])
1169            .expect("Configure to succeed in test");
1170        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1171        call_count.store(0, Ordering::SeqCst); // clear call count
1172
1173        s.establish().expect("Establish to succeed in test");
1174        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1175        call_count.store(0, Ordering::SeqCst); // clear call count
1176
1177        let (transport_chan, _remote) = create_test_channels(transport);
1178        assert_eq!(
1179            s.receive_channel(transport_chan).expect("Receive channel to succeed in test"),
1180            false
1181        );
1182        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1183        call_count.store(0, Ordering::SeqCst); // clear call count
1184
1185        s.start().expect("Start to succeed in test");
1186        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1187        call_count.store(0, Ordering::SeqCst); // clear call count
1188
1189        s.suspend().expect("Suspend to succeed in test");
1190        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1191        call_count.store(0, Ordering::SeqCst); // clear call count
1192
1193        s.reconfigure(vec![]).expect("Reconfigure to succeed in test");
1194        assert!(call_count.load(Ordering::SeqCst) > 0, "Update callback called at least once");
1195        call_count.store(0, Ordering::SeqCst); // clear call count
1196
1197        // Abort this stream, putting it back to the idle state.
1198        s.abort();
1199    }
1200}
bt_avdtp/stream_endpoint.rs

bt_avdtp/
stream_endpoint.rs
