| // Copyright 2019 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| use anyhow::{format_err, Error}; |
| use bt_avdtp as avdtp; |
| use fidl_fuchsia_bluetooth_bredr::{ |
| self as bredr, ChannelParameters, ProfileDescriptor, ProfileProxy, |
| }; |
| use fuchsia_async as fasync; |
| use fuchsia_bluetooth::{ |
| detachable_map::{DetachableMap, DetachableWeak}, |
| inspect::DebugExt, |
| types::{Channel, PeerId}, |
| }; |
| use fuchsia_inspect::{self as inspect, NumericProperty, Property}; |
| use fuchsia_inspect_derive::{AttachError, Inspect}; |
| use fuchsia_sync::Mutex; |
| use fuchsia_zircon as zx; |
| use futures::{ |
| channel::{mpsc, oneshot}, |
| stream::{Stream, StreamExt}, |
| task::{Context, Poll}, |
| Future, FutureExt, TryFutureExt, |
| }; |
| use std::{ |
| collections::{hash_map::Entry, HashMap, HashSet}, |
| pin::Pin, |
| sync::Arc, |
| }; |
| use tracing::{info, warn}; |
| |
| use crate::codec::CodecNegotiation; |
| use crate::stream::StreamsBuilder; |
| use crate::{peer::Peer, permits::Permits}; |
| |
| /// Statistics node for tracking various information about a peer that has been encountered. |
| /// Typically used as an inspect tree node. |
| struct PeerStats { |
| id: PeerId, |
| inspect_node: inspect::Node, |
| /// The number of times that this peer has been successfully connected to since discovery. |
| connection_count: inspect::UintProperty, |
| } |
| |
| impl PeerStats { |
| fn new(id: PeerId) -> Self { |
| Self { id, inspect_node: Default::default(), connection_count: Default::default() } |
| } |
| |
| fn set_descriptor(&mut self, descriptor: &ProfileDescriptor) { |
| self.inspect_node.record_string("descriptor", descriptor.debug()); |
| } |
| |
| fn record_connected(&mut self) { |
| let _ = self.connection_count.add(1); |
| } |
| } |
| |
| impl Inspect for &mut PeerStats { |
| fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> { |
| self.inspect_node = parent.create_child(name.as_ref()); |
| self.inspect_node.record_string("id", self.id.to_string()); |
| self.connection_count = self.inspect_node.create_uint("connection_count", 0); |
| Ok(()) |
| } |
| } |
| |
| #[derive(Default)] |
| struct DiscoveredPeers { |
| /// The peers that we have discovered, with their descriptors and potential preferred |
| /// endpoint directions. Because the same peer can be discovered multiple times, with |
| /// potentially different endpoints, we maintain a set of advertised directions. |
| descriptors: HashMap<PeerId, (ProfileDescriptor, HashSet<avdtp::EndpointType>)>, |
| /// Holds the child nodes which include the ids and profile descriptors for inspect. |
| stats: HashMap<PeerId, PeerStats>, |
| /// Inspect node, usually at "discovered" in the tree. |
| inspect_node: inspect::Node, |
| } |
| |
| impl DiscoveredPeers { |
| fn insert( |
| &mut self, |
| id: PeerId, |
| descriptor: ProfileDescriptor, |
| directions: HashSet<avdtp::EndpointType>, |
| ) { |
| match self.descriptors.entry(id) { |
| Entry::Occupied(mut entry) => { |
| entry.get_mut().0 = descriptor; |
| entry.get_mut().1.extend(&directions); |
| } |
| Entry::Vacant(entry) => { |
| let _ = entry.insert((descriptor, directions)); |
| } |
| }; |
| self.stats |
| .entry(id) |
| .or_insert({ |
| let mut new_stats = PeerStats::new(id); |
| let _ = new_stats.iattach(&self.inspect_node, inspect::unique_name("peer_")); |
| new_stats |
| }) |
| .set_descriptor(&descriptor); |
| } |
| |
| fn connected(&mut self, id: PeerId) { |
| if let Some(stats) = self.stats.get_mut(&id) { |
| stats.record_connected(); |
| } |
| } |
| |
| /// Returns the descriptor and preferred endpoint direction associated with the peer `id`. |
| fn get(&self, id: &PeerId) -> Option<(ProfileDescriptor, Option<avdtp::EndpointType>)> { |
| self.descriptors.get(id).map(|(desc, dirs)| (desc.clone(), find_preferred_direction(dirs))) |
| } |
| } |
| |
| impl Inspect for &mut DiscoveredPeers { |
| fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> { |
| self.inspect_node = parent.create_child(name.as_ref()); |
| Ok(()) |
| } |
| } |
| |
| /// Given a set of endpoint `directions`, returns the preferred direction or None |
| /// if both Sink and Source are specified. |
| fn find_preferred_direction( |
| directions: &HashSet<avdtp::EndpointType>, |
| ) -> Option<avdtp::EndpointType> { |
| if directions.len() == 1 { |
| directions.iter().next().cloned() |
| } else { |
| // Otherwise, either there are no A2DP services or both Sink & Source are specified |
| // in which case there is no preferred direction. |
| None |
| } |
| } |
| |
| /// Make an outgoing connection to a peer. |
| async fn connect_peer( |
| proxy: ProfileProxy, |
| id: PeerId, |
| channel_params: ChannelParameters, |
| ) -> Result<Channel, Error> { |
| info!(%id, "Connecting to peer"); |
| let connect_fut = proxy.connect( |
| &id.into(), |
| &bredr::ConnectParameters::L2cap(bredr::L2capParameters { |
| psm: Some(bredr::PSM_AVDTP), |
| parameters: Some(channel_params), |
| ..Default::default() |
| }), |
| ); |
| let channel = match connect_fut.await { |
| Err(e) => { |
| warn!(%id, ?e, "FIDL error on connect"); |
| return Err(e.into()); |
| } |
| Ok(Err(e)) => return Err(format_err!("Bluetooth connect error: {e:?}")), |
| Ok(Ok(channel)) => channel, |
| }; |
| |
| let channel = channel |
| .try_into() |
| .map_err(|e| format_err!("Couldn't convert FIDL to BT channel: {e:?}"))?; |
| Ok(channel) |
| } |
| |
| /// ConnectedPeers manages the set of connected peers based on discovery, new connection, and |
| /// peer session lifetime. |
| pub struct ConnectedPeers { |
| /// The set of connected peers. |
| connected: DetachableMap<PeerId, Peer>, |
| /// Tasks for peers that we are attempting to connect to. |
| /// Used to ensure only one outgoing attempt exists at once. |
| connection_attempts: Mutex<HashMap<PeerId, fasync::Task<()>>>, |
| /// ProfileDescriptors from discovering the peer, stored here even if the peer is disconnected |
| discovered: Mutex<DiscoveredPeers>, |
| /// Streams builder, provides a set of streams and negotiation when a peer is connected |
| streams_builder: StreamsBuilder, |
| /// The permits that each peer uses to validate that we can start a stream. |
| permits: Permits, |
| /// Profile Proxy, used to connect new transport sockets. |
| profile: ProfileProxy, |
| /// Cobalt logger to use and hand out to peers, if we are using one. |
| metrics: bt_metrics::MetricsLogger, |
| /// The 'peers' node of the inspect tree. All connected peers own a child node of this node. |
| inspect: inspect::Node, |
| /// Inspect node for which is the current preferred peer direction. |
| inspect_peer_direction: inspect::StringProperty, |
| /// Listeners for new connected peers |
| connected_peer_senders: Mutex<Vec<mpsc::Sender<DetachableWeak<PeerId, Peer>>>>, |
| /// Task handles for newly connected peer stream starts. |
| // TODO(https://fxbug.dev/42146917): Completed tasks aren't garbage-collected yet. |
| start_stream_tasks: Mutex<HashMap<PeerId, fasync::Task<()>>>, |
| /// Preferred direction for new peers. This is the direction we prefer the peer's endpoint to |
| /// be, i.e. if we prefer Sink, locally we are Source. |
| preferred_peer_direction: Mutex<avdtp::EndpointType>, |
| } |
| |
| impl ConnectedPeers { |
| pub fn new( |
| streams_builder: StreamsBuilder, |
| permits: Permits, |
| profile: ProfileProxy, |
| metrics: bt_metrics::MetricsLogger, |
| ) -> Self { |
| Self { |
| connected: DetachableMap::new(), |
| connection_attempts: Mutex::new(HashMap::new()), |
| discovered: Default::default(), |
| streams_builder, |
| profile, |
| permits, |
| inspect: inspect::Node::default(), |
| inspect_peer_direction: inspect::StringProperty::default(), |
| metrics, |
| connected_peer_senders: Default::default(), |
| start_stream_tasks: Default::default(), |
| preferred_peer_direction: Mutex::new(avdtp::EndpointType::Sink), |
| } |
| } |
| |
| pub(crate) fn get_weak(&self, id: &PeerId) -> Option<DetachableWeak<PeerId, Peer>> { |
| self.connected.get(id) |
| } |
| |
| pub(crate) fn get(&self, id: &PeerId) -> Option<Arc<Peer>> { |
| self.get_weak(id).and_then(|p| p.upgrade()) |
| } |
| |
| pub fn is_connected(&self, id: &PeerId) -> bool { |
| self.connected.contains_key(id) |
| } |
| |
| /// Attempts to start streaming on `peer` by collecting the remote streaming endpoint |
| /// information, selecting a compatible peer using `negotiation` and starting the stream. |
| /// Does nothing and returns Ok(()) if the peer is already streaming or will start streaming |
| /// on it's own. |
| async fn start_streaming( |
| peer: &DetachableWeak<PeerId, Peer>, |
| negotiation: CodecNegotiation, |
| ) -> Result<(), anyhow::Error> { |
| let remote_streams = { |
| let strong = peer.upgrade().ok_or(format_err!("Disconnected"))?; |
| if strong.streaming_active() { |
| return Ok(()); |
| } |
| strong.collect_capabilities() |
| } |
| .await?; |
| |
| let (negotiated, remote_seid) = |
| negotiation.select(&remote_streams).ok_or(format_err!("No compatible stream found"))?; |
| |
| let strong = peer.upgrade().ok_or(format_err!("Disconnected"))?; |
| if strong.streaming_active() { |
| let peer_id = peer.key(); |
| info!(%peer_id, "Not starting streaming, it's already started"); |
| return Ok(()); |
| } |
| strong.stream_start(remote_seid, negotiated).await.map_err(Into::into) |
| } |
| |
| pub fn found( |
| &self, |
| id: PeerId, |
| desc: ProfileDescriptor, |
| preferred_directions: HashSet<avdtp::EndpointType>, |
| ) { |
| self.discovered.lock().insert(id, desc.clone(), preferred_directions); |
| if let Some(peer) = self.get(&id) { |
| let _ = peer.set_descriptor(desc); |
| } |
| } |
| |
| pub fn set_preferred_peer_direction(&self, direction: avdtp::EndpointType) { |
| *self.preferred_peer_direction.lock() = direction; |
| self.inspect_peer_direction.set(&format!("{direction:?}")); |
| } |
| |
| pub fn preferred_peer_direction(&self) -> avdtp::EndpointType { |
| *self.preferred_peer_direction.lock() |
| } |
| |
| pub fn try_connect( |
| &self, |
| id: PeerId, |
| channel_params: ChannelParameters, |
| ) -> impl Future<Output = Result<Option<Channel>, Error>> { |
| let proxy = self.profile.clone(); |
| let connected = self.is_connected(&id); |
| let (sender, recv) = oneshot::channel(); |
| let recv = |
| recv.map_ok_or_else(|_e| Err(format_err!("Connection task canceled")), Into::into); |
| if connected { |
| if let Err(e) = sender.send(Ok(None)) { |
| warn!(%id, ?e, "Failed to notify already-connected"); |
| } |
| return recv; |
| } |
| let mut attempts = self.connection_attempts.lock(); |
| if let Some(previous_connect_task) = attempts.remove(&id) { |
| // We are the only place that can poll the connect task, check if it finished. |
| if previous_connect_task.now_or_never().is_none() { |
| warn!(%id, "Cancelling previous connect attempt"); |
| } |
| } |
| let connect_task = fasync::Task::spawn(async move { |
| if let Err(e) = sender.send(connect_peer(proxy, id, channel_params).await.map(Some)) { |
| warn!(%id, ?e, "Failed to send channel connect result"); |
| } |
| }); |
| let _ = attempts.insert(id, connect_task); |
| recv |
| } |
| |
| /// Accept a channel that is connected to the peer `id`. |
| /// If `initiator_delay` is set, attempt to start a stream after the specified delay. |
| /// `initiator_delay` has no effect if the peer already has a control channel. |
| /// Returns a weak peer pointer (even if it was previously connected) if successful. |
| pub async fn connected( |
| &self, |
| id: PeerId, |
| channel: Channel, |
| initiator_delay: Option<zx::Duration>, |
| ) -> Result<DetachableWeak<PeerId, Peer>, Error> { |
| if let Some(weak) = self.get_weak(&id) { |
| let peer = weak.upgrade().ok_or(format_err!("Disconnected connecting transport"))?; |
| if let Err(e) = peer.receive_channel(channel) { |
| warn!(%id, %e, "failed to connect channel"); |
| return Err(e.into()); |
| } |
| return Ok(weak); |
| } |
| |
| let entry = self.connected.lazy_entry(&id); |
| |
| info!(%id, "peer connected"); |
| let avdtp_peer = avdtp::Peer::new(channel); |
| |
| let mut peer = Peer::create( |
| id, |
| avdtp_peer, |
| self.streams_builder.peer_streams(&id, None).await?, |
| Some(self.permits.clone()), |
| self.profile.clone(), |
| self.metrics.clone(), |
| ); |
| |
| self.discovered.lock().connected(id); |
| |
| let peer_preferred_direction = if let Some((desc, dir)) = self.discovered.lock().get(&id) { |
| let _ = peer.set_descriptor(desc); |
| dir |
| } else { |
| None |
| }; |
| |
| if let Err(e) = peer.iattach(&self.inspect, inspect::unique_name("peer_")) { |
| warn!(%id, ?e, "Couldn't attach inspect"); |
| } |
| |
| let closed_fut = peer.closed(); |
| let peer = match entry.try_insert(peer) { |
| Err(_peer) => { |
| warn!(%id, "Peer connected while we were setting up"); |
| return self.get_weak(&id).ok_or(format_err!("Peer missing")); |
| } |
| Ok(weak_peer) => weak_peer, |
| }; |
| |
| if let Some(delay) = initiator_delay { |
| let peer = peer.clone(); |
| let peer_id = peer.key().clone(); |
| // Bias the codec negotiation with the peer's preferred direction that was discovered |
| // from the SDP service search. |
| let negotiation = self |
| .streams_builder |
| .negotiation( |
| &id, |
| None, |
| peer_preferred_direction.unwrap_or(self.preferred_peer_direction()), |
| ) |
| .await?; |
| let start_stream_task = fuchsia_async::Task::local(async move { |
| let delay_sec = delay.into_millis() as f64 / 1000.0; |
| info!(id = %peer.key(), "dwelling {delay_sec}s for peer initiation"); |
| fasync::Timer::new(fasync::Time::after(delay)).await; |
| |
| if let Err(e) = ConnectedPeers::start_streaming(&peer, negotiation).await { |
| info!(id = %peer.key(), ?e, "Peer start streaming failed"); |
| peer.detach(); |
| } |
| }); |
| if self.start_stream_tasks.lock().insert(peer_id, start_stream_task).is_some() { |
| info!(%peer_id, "Replacing previous start stream dwell"); |
| } |
| } |
| |
| // Remove the peer when we disconnect. |
| fasync::Task::local(async move { |
| closed_fut.await; |
| peer.detach(); |
| }) |
| .detach(); |
| |
| let peer = self.get_weak(&id).ok_or(format_err!("Peer missing"))?; |
| self.notify_connected(&peer); |
| Ok(peer) |
| } |
| |
| /// Notify the listeners that a new peer has been connected to. |
| fn notify_connected(&self, peer: &DetachableWeak<PeerId, Peer>) { |
| let mut senders = self.connected_peer_senders.lock(); |
| senders.retain_mut(|sender| sender.try_send(peer.clone()).is_ok()); |
| } |
| |
| /// Get a stream that produces peers that have been connected. |
| pub fn connected_stream(&self) -> PeerConnections { |
| let (sender, receiver) = mpsc::channel(0); |
| self.connected_peer_senders.lock().push(sender); |
| PeerConnections { stream: receiver } |
| } |
| } |
| |
| impl Inspect for &mut ConnectedPeers { |
| fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> { |
| self.inspect = parent.create_child(name.as_ref()); |
| let peer_dir_str = format!("{:?}", self.preferred_peer_direction()); |
| self.inspect_peer_direction = |
| self.inspect.create_string("preferred_peer_direction", peer_dir_str); |
| self.streams_builder.iattach(&self.inspect, "streams_builder")?; |
| self.discovered.lock().iattach(&self.inspect, "discovered") |
| } |
| } |
| |
| /// Provides a stream of peers that have been connected to. This stream produces an item whenever |
| /// an A2DP peer has been connected. It will produce None when no more peers will be connected. |
| pub struct PeerConnections { |
| stream: mpsc::Receiver<DetachableWeak<PeerId, Peer>>, |
| } |
| |
| impl Stream for PeerConnections { |
| type Item = DetachableWeak<PeerId, Peer>; |
| |
| fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| self.stream.poll_next_unpin(cx) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| |
| use async_utils::PollExt; |
| use bt_avdtp::{Request, ServiceCapability}; |
| use diagnostics_assertions::assert_data_tree; |
| use fidl::endpoints::create_proxy_and_stream; |
| use fidl_fuchsia_bluetooth_bredr::{ |
| AudioOffloadExtProxy, ProfileMarker, ProfileRequestStream, ServiceClassProfileIdentifier, |
| }; |
| use futures::future::BoxFuture; |
| use std::pin::pin; |
| |
| use crate::codec::MediaCodecConfig; |
| use crate::media_task::{MediaTaskBuilder, MediaTaskError, MediaTaskRunner}; |
| use crate::media_types::*; |
| |
| fn run_to_stalled(exec: &mut fasync::TestExecutor) { |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| } |
| |
| fn exercise_avdtp(exec: &mut fasync::TestExecutor, remote: Channel, peer: &Peer) { |
| let remote_avdtp = avdtp::Peer::new(remote); |
| let mut remote_requests = remote_avdtp.take_request_stream(); |
| |
| // Should be able to actually communicate via the peer. |
| let avdtp = peer.avdtp(); |
| let discover_fut = avdtp.discover(); |
| |
| let mut discover_fut = pin!(discover_fut); |
| |
| assert!(exec.run_until_stalled(&mut discover_fut).is_pending()); |
| |
| let responder = match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(Request::Discover { responder }))) => responder, |
| x => panic!("Expected a Ready Discovery request but got {:?}", x), |
| }; |
| |
| let endpoint_id = avdtp::StreamEndpointId::try_from(1).expect("endpointid creation"); |
| |
| let information = avdtp::StreamInformation::new( |
| endpoint_id, |
| false, |
| avdtp::MediaType::Audio, |
| avdtp::EndpointType::Source, |
| ); |
| |
| responder.send(&[information]).expect("Sending response should have worked"); |
| |
| let _stream_infos = match exec.run_until_stalled(&mut discover_fut) { |
| Poll::Ready(Ok(infos)) => infos, |
| x => panic!("Expected a Ready response but got {:?}", x), |
| }; |
| } |
| |
| fn setup_connected_peer_test( |
| ) -> (fasync::TestExecutor, PeerId, ConnectedPeers, ProfileRequestStream) { |
| let exec = fasync::TestExecutor::new(); |
| let (proxy, stream) = |
| create_proxy_and_stream::<ProfileMarker>().expect("Profile proxy should be created"); |
| let id = PeerId(1); |
| |
| let peers = ConnectedPeers::new( |
| StreamsBuilder::default(), |
| Permits::new(1), |
| proxy, |
| bt_metrics::MetricsLogger::default(), |
| ); |
| |
| (exec, id, peers, stream) |
| } |
| |
| #[fuchsia::test] |
| fn connect_creates_peer() { |
| let (mut exec, id, peers, _stream) = setup_connected_peer_test(); |
| |
| let (remote, channel) = Channel::create(); |
| |
| let peer = exec |
| .run_singlethreaded(peers.connected(id, channel, None)) |
| .expect("peer should connect"); |
| let peer = peer.upgrade().expect("peer should be connected"); |
| |
| exercise_avdtp(&mut exec, remote, &peer); |
| } |
| |
| #[fuchsia::test] |
| fn connect_notifies_streams() { |
| let (mut exec, id, peers, _stream) = setup_connected_peer_test(); |
| |
| let (remote, channel) = Channel::create(); |
| |
| let mut peer_stream = peers.connected_stream(); |
| let mut peer_stream_two = peers.connected_stream(); |
| |
| let peer = exec |
| .run_singlethreaded(peers.connected(id, channel, None)) |
| .expect("peer should connect"); |
| let peer = peer.upgrade().expect("peer should be connected"); |
| |
| // Peers should have been notified of the new peer |
| let weak = exec.run_singlethreaded(peer_stream.next()).expect("peer stream to produce"); |
| assert_eq!(weak.key(), &id); |
| let weak = exec.run_singlethreaded(peer_stream_two.next()).expect("peer stream to produce"); |
| assert_eq!(weak.key(), &id); |
| |
| exercise_avdtp(&mut exec, remote, &peer); |
| |
| // If you drop one stream, the other one should still produce. |
| drop(peer_stream); |
| |
| let id2 = PeerId(2); |
| let (remote2, channel2) = Channel::create(); |
| let peer2 = exec |
| .run_singlethreaded(peers.connected(id2, channel2, None)) |
| .expect("peer should connect"); |
| let peer2 = peer2.upgrade().expect("peer two should be connected"); |
| |
| let weak = exec.run_singlethreaded(peer_stream_two.next()).expect("peer stream to produce"); |
| assert_eq!(weak.key(), &id2); |
| |
| exercise_avdtp(&mut exec, remote2, &peer2); |
| } |
| |
| #[fuchsia::test] |
| fn find_preferred_direction_returns_correct_endpoints() { |
| let empty = HashSet::new(); |
| assert_eq!(find_preferred_direction(&empty), None); |
| |
| let sink_only = HashSet::from_iter(vec![avdtp::EndpointType::Sink].into_iter()); |
| assert_eq!(find_preferred_direction(&sink_only), Some(avdtp::EndpointType::Sink)); |
| |
| let source_only = HashSet::from_iter(vec![avdtp::EndpointType::Source].into_iter()); |
| assert_eq!(find_preferred_direction(&source_only), Some(avdtp::EndpointType::Source)); |
| |
| let both = HashSet::from_iter( |
| vec![avdtp::EndpointType::Sink, avdtp::EndpointType::Source].into_iter(), |
| ); |
| assert_eq!(find_preferred_direction(&both), None); |
| } |
| |
| // Expected chosen ID for the AAC stream endpoint. |
| const AAC_SEID: u8 = 8; |
| // Expected chosen ID for the SBC sink stream endpoint. |
| const SBC_SINK_SEID: u8 = 9; |
| // Expected chosen ID for the SBC source stream endpoint. |
| const SBC_SOURCE_SEID: u8 = 10; |
| |
| fn aac_sink_codec() -> avdtp::ServiceCapability { |
| AacCodecInfo::new( |
| AacObjectType::MANDATORY_SNK, |
| AacSamplingFrequency::MANDATORY_SNK, |
| AacChannels::MANDATORY_SNK, |
| true, |
| 0, // 0 = Unknown constant bitrate support (A2DP Sec. 4.5.2.4) |
| ) |
| .unwrap() |
| .into() |
| } |
| |
| fn sbc_sink_codec() -> avdtp::ServiceCapability { |
| SbcCodecInfo::new( |
| SbcSamplingFrequency::MANDATORY_SNK, |
| SbcChannelMode::MANDATORY_SNK, |
| SbcBlockCount::MANDATORY_SNK, |
| SbcSubBands::MANDATORY_SNK, |
| SbcAllocation::MANDATORY_SNK, |
| SbcCodecInfo::BITPOOL_MIN, |
| SbcCodecInfo::BITPOOL_MAX, |
| ) |
| .unwrap() |
| .into() |
| } |
| |
| fn sbc_source_codec() -> avdtp::ServiceCapability { |
| SbcCodecInfo::new( |
| SbcSamplingFrequency::FREQ48000HZ, |
| SbcChannelMode::JOINT_STEREO, |
| SbcBlockCount::MANDATORY_SRC, |
| SbcSubBands::MANDATORY_SRC, |
| SbcAllocation::MANDATORY_SRC, |
| SbcCodecInfo::BITPOOL_MIN, |
| SbcCodecInfo::BITPOOL_MAX, |
| ) |
| .unwrap() |
| .into() |
| } |
| |
| #[derive(Clone)] |
| struct FakeBuilder { |
| capability: avdtp::ServiceCapability, |
| direction: avdtp::EndpointType, |
| } |
| |
| impl MediaTaskBuilder for FakeBuilder { |
| fn configure( |
| &self, |
| _peer_id: &PeerId, |
| codec_config: &MediaCodecConfig, |
| ) -> Result<Box<dyn MediaTaskRunner>, MediaTaskError> { |
| if self.capability.codec_type() == Some(codec_config.codec_type()) { |
| return Ok(Box::new(FakeRunner {})); |
| } |
| Err(MediaTaskError::Other(String::from("Unsupported configuring"))) |
| } |
| |
| fn direction(&self) -> bt_avdtp::EndpointType { |
| self.direction |
| } |
| |
| fn supported_configs( |
| &self, |
| _peer_id: &PeerId, |
| _offload: Option<AudioOffloadExtProxy>, |
| ) -> BoxFuture<'static, Result<Vec<MediaCodecConfig>, MediaTaskError>> { |
| futures::future::ready(Ok(vec![(&self.capability).try_into().unwrap()])).boxed() |
| } |
| } |
| |
| struct FakeRunner {} |
| |
| impl MediaTaskRunner for FakeRunner { |
| fn start( |
| &mut self, |
| _stream: avdtp::MediaStream, |
| _offload: Option<AudioOffloadExtProxy>, |
| ) -> Result<Box<dyn crate::media_task::MediaTask>, MediaTaskError> { |
| Err(MediaTaskError::Other(String::from("unimplemented starting"))) |
| } |
| } |
| |
| /// Sets up a test in which we expect to select a stream and connect to a peer. |
| /// Returns the executor, connected peers (under test), request stream for profile interaction, |
| /// and an SBC and AAC Sink service capability. |
| fn setup_negotiation_test() -> ( |
| fasync::TestExecutor, |
| ConnectedPeers, |
| ProfileRequestStream, |
| ServiceCapability, |
| ServiceCapability, |
| ) { |
| let exec = fasync::TestExecutor::new_with_fake_time(); |
| exec.set_fake_time(fasync::Time::from_nanos(1_000_000)); |
| let (proxy, stream) = |
| create_proxy_and_stream::<ProfileMarker>().expect("Profile proxy should be created"); |
| |
| let aac_sink_codec = aac_sink_codec(); |
| let sbc_sink_codec = sbc_sink_codec(); |
| let aac_sink_builder = FakeBuilder { |
| capability: aac_sink_codec.clone(), |
| direction: avdtp::EndpointType::Sink, |
| }; |
| let sbc_sink_builder = FakeBuilder { |
| capability: sbc_sink_codec.clone(), |
| direction: avdtp::EndpointType::Sink, |
| }; |
| let sbc_source_builder = |
| FakeBuilder { capability: sbc_source_codec(), direction: avdtp::EndpointType::Source }; |
| |
| let mut streams_builder = StreamsBuilder::default(); |
| streams_builder.add_builder(aac_sink_builder); |
| streams_builder.add_builder(sbc_sink_builder); |
| streams_builder.add_builder(sbc_source_builder); |
| |
| let peers = ConnectedPeers::new( |
| streams_builder, |
| Permits::new(1), |
| proxy, |
| bt_metrics::MetricsLogger::default(), |
| ); |
| |
| (exec, peers, stream, sbc_sink_codec, aac_sink_codec) |
| } |
| |
| #[fuchsia::test] |
| fn streaming_start_with_streaming_peer_is_noop() { |
| let (mut exec, peers, _stream, sbc_codec, _aac_codec) = setup_negotiation_test(); |
| let id = PeerId(1); |
| let (remote, channel) = Channel::create(); |
| let remote = avdtp::Peer::new(remote); |
| |
| let delay = zx::Duration::from_seconds(1); |
| |
| let mut remote_requests = remote.take_request_stream(); |
| |
| // This starts the task in the background waiting. |
| let mut connected_fut = std::pin::pin!(peers.connected(id, channel, Some(delay))); |
| assert!(exec.run_until_stalled(&mut connected_fut).expect("ready").is_ok()); |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| // Before the delay expires, the peer starts the stream. |
| |
| let seid: avdtp::StreamEndpointId = SBC_SINK_SEID.try_into().expect("seid to be okay"); |
| let config_caps = &[ServiceCapability::MediaTransport, sbc_codec]; |
| let set_config_fut = remote.set_configuration(&seid, &seid, config_caps); |
| let mut set_config_fut = pin!(set_config_fut); |
| match exec.run_until_stalled(&mut set_config_fut) { |
| Poll::Ready(Ok(())) => {} |
| x => panic!("Expected set config to be ready and Ok, got {:?}", x), |
| }; |
| |
| // The remote peer doesn't need to actually open, Set Configuration is enough of a signal. |
| // wait for the delay to expire now. |
| |
| exec.set_fake_time(fasync::Time::after(delay) + zx::Duration::from_micros(1)); |
| let _ = exec.wake_expired_timers(); |
| |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| // Shouldn't start a discovery, since the stream is scheduled to start already. |
| assert!(exec.run_until_stalled(&mut remote_requests.next()).is_pending()); |
| } |
| |
| fn sbc_source_endpoint() -> (avdtp::StreamEndpointId, avdtp::StreamInformation) { |
| let remote_sbc_seid: avdtp::StreamEndpointId = 1u8.try_into().unwrap(); |
| let info = avdtp::StreamInformation::new( |
| remote_sbc_seid.clone(), |
| false, |
| avdtp::MediaType::Audio, |
| avdtp::EndpointType::Source, |
| ); |
| (remote_sbc_seid, info) |
| } |
| |
| fn aac_source_endpoint() -> (avdtp::StreamEndpointId, avdtp::StreamInformation) { |
| let remote_aac_seid: avdtp::StreamEndpointId = 2u8.try_into().unwrap(); |
| let info = avdtp::StreamInformation::new( |
| remote_aac_seid.clone(), |
| false, |
| avdtp::MediaType::Audio, |
| avdtp::EndpointType::Source, |
| ); |
| (remote_aac_seid, info) |
| } |
| |
| fn sbc_sink_endpoint() -> (avdtp::StreamEndpointId, avdtp::StreamInformation) { |
| let remote_sbc_seid: avdtp::StreamEndpointId = 3u8.try_into().unwrap(); |
| let info = avdtp::StreamInformation::new( |
| remote_sbc_seid.clone(), |
| false, |
| avdtp::MediaType::Audio, |
| avdtp::EndpointType::Sink, |
| ); |
| (remote_sbc_seid, info) |
| } |
| |
| /// Expects an AVDTP Discovery request on the `requests` stream. Responds to |
| /// the request with the provided `response` endpoints. |
| fn expect_peer_discovery( |
| exec: &mut fasync::TestExecutor, |
| requests: &mut avdtp::RequestStream, |
| response: Vec<avdtp::StreamInformation>, |
| ) { |
| match exec.run_until_stalled(&mut requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::Discover { responder }))) => { |
| responder.send(&response).expect("response succeeds"); |
| } |
| x => panic!("Expected a discovery request to be sent after delay, got {:?}", x), |
| }; |
| } |
| |
| #[fuchsia::test] |
| fn streaming_start_configure_while_discovery() { |
| let (mut exec, peers, _stream, sbc_codec, _aac_codec) = setup_negotiation_test(); |
| let id = PeerId(1); |
| let (remote, channel) = Channel::create(); |
| let remote = avdtp::Peer::new(remote); |
| |
| let delay = zx::Duration::from_seconds(1); |
| |
| let mut remote_requests = remote.take_request_stream(); |
| |
| // This starts the task in the background waiting. |
| let mut connected_fut = std::pin::pin!(peers.connected(id, channel, Some(delay))); |
| assert!(exec.run_until_stalled(&mut connected_fut).expect("ready").is_ok()); |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| // The delay expires, and the discovery is start! |
| exec.set_fake_time(fasync::Time::after(delay) + zx::Duration::from_micros(1)); |
| let _ = exec.wake_expired_timers(); |
| expect_peer_discovery( |
| &mut exec, |
| &mut remote_requests, |
| vec![sbc_source_endpoint().1, aac_source_endpoint().1], |
| ); |
| |
| // The remote peer doesn't need to actually open, Set Configuration is enough of a signal. |
| let seid: avdtp::StreamEndpointId = SBC_SINK_SEID.try_into().expect("seid to be okay"); |
| let config_caps = &[ServiceCapability::MediaTransport, sbc_codec.clone()]; |
| let set_config_fut = remote.set_configuration(&seid, &seid, config_caps); |
| let mut set_config_fut = pin!(set_config_fut); |
| match exec.run_until_stalled(&mut set_config_fut) { |
| Poll::Ready(Ok(())) => {} |
| x => panic!("Expected set config to be ready and Ok, got {:?}", x), |
| }; |
| |
| // Can finish the collection process, but not attempt to configure or start a stream. |
| loop { |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::GetCapabilities { responder, .. }))) => { |
| responder |
| .send(&[avdtp::ServiceCapability::MediaTransport, sbc_codec.clone()]) |
| .expect("respond succeeds"); |
| } |
| Poll::Ready(x) => panic!("Got unexpected request: {:?}", x), |
| Poll::Pending => break, |
| } |
| } |
| } |
| |
| /// Tests connection initiation selects the appropriate stream endpoint based |
| /// on a biased codec negotiation that is set from the peer's discovered services. |
| #[fuchsia::test] |
| fn connect_initiation_uses_biased_codec_negotiation_by_peer() { |
| let (mut exec, peers, _stream, sbc_codec, _aac_codec) = setup_negotiation_test(); |
| let id = PeerId(1); |
| let (remote, channel) = Channel::create(); |
| |
| // System biases towards the Source direction (called when the AudioMode FIDL changes). |
| peers.set_preferred_peer_direction(avdtp::EndpointType::Source); |
| |
| // New fake peer discovered with some descriptor - the peer's SDP entry shows Sink. |
| let remote = avdtp::Peer::new(remote); |
| let desc = ProfileDescriptor { |
| profile_id: ServiceClassProfileIdentifier::AdvancedAudioDistribution, |
| major_version: 1, |
| minor_version: 2, |
| }; |
| let preferred_direction = vec![avdtp::EndpointType::Sink]; |
| let delay = zx::Duration::from_seconds(1); |
| peers.found(id, desc, HashSet::from_iter(preferred_direction.into_iter())); |
| |
| let connected_fut = peers.connected(id, channel, Some(delay)); |
| let mut connected_fut = std::pin::pin!(connected_fut); |
| let _ = exec |
| .run_until_stalled(&mut connected_fut) |
| .expect("is ready") |
| .expect("connect control channel is ok"); |
| // run the start task until it's stalled. |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| let mut remote_requests = remote.take_request_stream(); |
| |
| // Should wait for the specified amount of time. |
| assert!(exec.run_until_stalled(&mut remote_requests.next()).is_pending()); |
| |
| exec.set_fake_time(fasync::Time::after(delay + zx::Duration::from_micros(1))); |
| let _ = exec.wake_expired_timers(); |
| |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| // Even though the peer supports both SBC Sink and Source, we expect to negotiate and start |
| // on the Sink endpoint since that is the peer's preferred one. |
| let (peer_sbc_source_seid, peer_sbc_source_endpoint) = sbc_source_endpoint(); |
| let (peer_sbc_sink_seid, peer_sbc_sink_endpoint) = sbc_sink_endpoint(); |
| expect_peer_discovery( |
| &mut exec, |
| &mut remote_requests, |
| vec![peer_sbc_source_endpoint, peer_sbc_sink_endpoint], |
| ); |
| for _twice in 1..=2 { |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::GetCapabilities { stream_id, responder }))) => { |
| let codec = match stream_id { |
| id if id == peer_sbc_source_seid => sbc_codec.clone(), |
| id if id == peer_sbc_sink_seid => sbc_codec.clone(), |
| x => panic!("Got unexpected get_capabilities seid {:?}", x), |
| }; |
| responder |
| .send(&[avdtp::ServiceCapability::MediaTransport, codec]) |
| .expect("respond succeeds"); |
| } |
| x => panic!("Expected a ready get capabilities request, got {:?}", x), |
| }; |
| } |
| |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::SetConfiguration { |
| local_stream_id, |
| remote_stream_id, |
| capabilities: _, |
| responder, |
| }))) => { |
| // We expect the set configuration to apply to the remote peer's Sink SEID and the |
| // local Source SEID. |
| assert_eq!(peer_sbc_sink_seid, local_stream_id); |
| let local_sbc_source_seid: avdtp::StreamEndpointId = |
| SBC_SOURCE_SEID.try_into().unwrap(); |
| assert_eq!(local_sbc_source_seid, remote_stream_id); |
| responder.send().expect("response sends"); |
| } |
| x => panic!("Expected a ready set configuration request, got {:?}", x), |
| }; |
| } |
| |
| /// Tests connection initiation selects the appropriate stream endpoint based |
| /// on a biased codec negotiation that is set from by the system (in practice, the AudioMode |
| /// FIDL). This case typically occurs when a peer advertises both sink and source, and therefore |
| /// has no preference for the endpoint direction. |
| #[fuchsia::test] |
| fn connect_initiation_uses_biased_codec_negotiation_by_system() { |
| let (mut exec, peers, _stream, sbc_codec, _aac_codec) = setup_negotiation_test(); |
| let id = PeerId(1); |
| let (remote, channel) = Channel::create(); |
| |
| // System biases towards the Source direction (called when the AudioMode FIDL changes). |
| peers.set_preferred_peer_direction(avdtp::EndpointType::Source); |
| |
| // New fake peer discovered with separate Sink and Source entries. |
| let remote = avdtp::Peer::new(remote); |
| let desc = ProfileDescriptor { |
| profile_id: ServiceClassProfileIdentifier::AdvancedAudioDistribution, |
| major_version: 1, |
| minor_version: 2, |
| }; |
| peers.found( |
| id, |
| desc.clone(), |
| HashSet::from_iter(vec![avdtp::EndpointType::Source].into_iter()), |
| ); |
| peers.found(id, desc, HashSet::from_iter(vec![avdtp::EndpointType::Sink].into_iter())); |
| |
| let delay = zx::Duration::from_seconds(1); |
| let connect_fut = peers.connected(id, channel, Some(delay)); |
| let mut connect_fut = std::pin::pin!(connect_fut); |
| let _ = exec |
| .run_until_stalled(&mut connect_fut) |
| .expect("ready") |
| .expect("connect control channel is ok"); |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| let mut remote_requests = remote.take_request_stream(); |
| // Should wait for the specified amount of time. |
| assert!(exec.run_until_stalled(&mut remote_requests.next()).is_pending()); |
| exec.set_fake_time(fasync::Time::after(delay + zx::Duration::from_micros(1))); |
| let _ = exec.wake_expired_timers(); |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| // Because the peer advertises both Sink and Source, we fall back to the system-biased |
| // direction, which is Source for the peer. |
| let (peer_sbc_source_seid, peer_sbc_source_endpoint) = sbc_source_endpoint(); |
| let (peer_sbc_sink_seid, peer_sbc_sink_endpoint) = sbc_sink_endpoint(); |
| expect_peer_discovery( |
| &mut exec, |
| &mut remote_requests, |
| vec![peer_sbc_source_endpoint, peer_sbc_sink_endpoint], |
| ); |
| for _twice in 1..=2 { |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::GetCapabilities { stream_id, responder }))) => { |
| let codec = match stream_id { |
| id if id == peer_sbc_source_seid => sbc_codec.clone(), |
| id if id == peer_sbc_sink_seid => sbc_codec.clone(), |
| x => panic!("Got unexpected get_capabilities seid {:?}", x), |
| }; |
| responder |
| .send(&[avdtp::ServiceCapability::MediaTransport, codec]) |
| .expect("respond succeeds"); |
| } |
| x => panic!("Expected a ready get capabilities request, got {:?}", x), |
| }; |
| } |
| |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::SetConfiguration { |
| local_stream_id, |
| remote_stream_id, |
| capabilities: _, |
| responder, |
| }))) => { |
| // We expect the set configuration to apply to the remote peer's Source SEID and the |
| // local Sink SEID. |
| assert_eq!(peer_sbc_source_seid, local_stream_id); |
| let local_sbc_sink_seid: avdtp::StreamEndpointId = |
| SBC_SINK_SEID.try_into().unwrap(); |
| assert_eq!(local_sbc_sink_seid, remote_stream_id); |
| responder.send().expect("response sends"); |
| } |
| x => panic!("Expected a ready set configuration request, got {:?}", x), |
| }; |
| } |
| |
| #[fuchsia::test] |
| fn connect_initiation_uses_negotiation() { |
| let (mut exec, peers, _stream, sbc_codec, aac_codec) = setup_negotiation_test(); |
| let id = PeerId(1); |
| let (remote, channel) = Channel::create(); |
| let remote = avdtp::Peer::new(remote); |
| |
| let delay = zx::Duration::from_seconds(1); |
| |
| let mut connect_fut = std::pin::pin!(peers.connected(id, channel, Some(delay))); |
| let _ = exec |
| .run_until_stalled(&mut connect_fut) |
| .expect("ready") |
| .expect("connect control channel is ok"); |
| |
| // run the start task until it's stalled. |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| let mut remote_requests = remote.take_request_stream(); |
| |
| // Should wait for the specified amount of time. |
| assert!(exec.run_until_stalled(&mut remote_requests.next()).is_pending()); |
| |
| exec.set_fake_time(fasync::Time::after(delay + zx::Duration::from_micros(1))); |
| let _ = exec.wake_expired_timers(); |
| |
| let _ = exec.run_until_stalled(&mut futures::future::pending::<()>()); |
| |
| // Should discover remote streams, negotiate, and start. |
| let (peer_sbc_seid, peer_sbc_endpoint) = sbc_source_endpoint(); |
| let (peer_aac_seid, peer_aac_endpoint) = aac_source_endpoint(); |
| expect_peer_discovery( |
| &mut exec, |
| &mut remote_requests, |
| vec![peer_sbc_endpoint, peer_aac_endpoint], |
| ); |
| for _twice in 1..=2 { |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::GetCapabilities { stream_id, responder }))) => { |
| let codec = match stream_id { |
| id if id == peer_sbc_seid => sbc_codec.clone(), |
| id if id == peer_aac_seid => aac_codec.clone(), |
| x => panic!("Got unexpected get_capabilities seid {:?}", x), |
| }; |
| responder |
| .send(&[avdtp::ServiceCapability::MediaTransport, codec]) |
| .expect("respond succeeds"); |
| } |
| x => panic!("Expected a ready get capabilities request, got {:?}", x), |
| }; |
| } |
| |
| match exec.run_until_stalled(&mut remote_requests.next()) { |
| Poll::Ready(Some(Ok(avdtp::Request::SetConfiguration { |
| local_stream_id, |
| remote_stream_id, |
| capabilities: _, |
| responder, |
| }))) => { |
| // Should set the aac stream, matched with local AAC seid. |
| assert_eq!(peer_aac_seid, local_stream_id); |
| let local_aac_seid: avdtp::StreamEndpointId = AAC_SEID.try_into().unwrap(); |
| assert_eq!(local_aac_seid, remote_stream_id); |
| responder.send().expect("response sends"); |
| } |
| x => panic!("Expected a ready set configuration request, got {:?}", x), |
| }; |
| } |
| |
| #[fuchsia::test] |
| fn connected_peers_inspect() { |
| let (mut exec, id, mut peers, _stream) = setup_connected_peer_test(); |
| |
| let inspect = inspect::Inspector::default(); |
| peers.iattach(inspect.root(), "peers").expect("should attach to inspect tree"); |
| |
| assert_data_tree!(inspect, root: { |
| peers: { streams_builder: contains {}, discovered: contains {}, preferred_peer_direction: "Sink" }}); |
| |
| peers.set_preferred_peer_direction(avdtp::EndpointType::Source); |
| |
| assert_data_tree!(inspect, root: { |
| peers: { streams_builder: contains {}, discovered: contains {}, preferred_peer_direction: "Source" }}); |
| |
| // Connect a peer, it should show up in the tree. |
| let (_remote, channel) = Channel::create(); |
| assert!(exec.run_singlethreaded(peers.connected(id, channel, None)).is_ok()); |
| |
| assert_data_tree!(inspect, root: { |
| peers: { |
| discovered: contains {}, |
| preferred_peer_direction: "Source", |
| streams_builder: contains {}, |
| peer_0: { id: "0000000000000001", local_streams: contains {} } |
| } |
| }); |
| } |
| |
| #[fuchsia::test] |
| fn try_connect_cancels_previous_attempt() { |
| let (mut exec, id, peers, mut profile_stream) = setup_connected_peer_test(); |
| |
| let mut connect_fut = peers.try_connect(id, ChannelParameters::default()); |
| |
| // Should get a request to connect, which we will stall and not respond to. |
| let responder = match exec.run_singlethreaded(profile_stream.next()) { |
| Some(Ok(bredr::ProfileRequest::Connect { responder, .. })) => responder, |
| x => panic!("Expected Profile connect, got {x:?}"), |
| }; |
| |
| // Trying to connect again should cancel the first try, and send another connect. |
| let mut connect_again_fut = peers.try_connect(id, ChannelParameters::default()); |
| let responder_two = match exec.run_singlethreaded(profile_stream.next()) { |
| Some(Ok(bredr::ProfileRequest::Connect { responder, .. })) => responder, |
| x => panic!("Expected Profile connect, got {x:?}"), |
| }; |
| |
| let first_result = exec.run_singlethreaded(&mut connect_fut); |
| let _ = first_result.expect_err("Should have an error from first attempt"); |
| |
| // Responding on the first connect shouldn't do anything at this point. |
| responder.send(Err(fidl_fuchsia_bluetooth::ErrorCode::Failed)).unwrap(); |
| |
| exec.run_until_stalled(&mut connect_again_fut).expect_pending("shouldn't finish"); |
| |
| let (_remote, local) = Channel::create(); |
| responder_two.send(Ok(local.try_into().unwrap())).unwrap(); |
| |
| let second_result = exec.run_singlethreaded(&mut connect_again_fut); |
| let _ = second_result.expect("should receive the channel"); |
| } |
| |
| #[fuchsia::test] |
| fn connected_peers_peer_disconnect_removes_peer() { |
| let (mut exec, id, peers, _stream) = setup_connected_peer_test(); |
| |
| let (remote, channel) = Channel::create(); |
| |
| assert!(exec.run_singlethreaded(peers.connected(id, channel, None)).is_ok()); |
| run_to_stalled(&mut exec); |
| |
| // Disconnect the signaling channel, peer should be gone. |
| drop(remote); |
| |
| run_to_stalled(&mut exec); |
| |
| assert!(peers.get(&id).is_none()); |
| } |
| |
| #[fuchsia::test] |
| fn connected_peers_reconnect_works() { |
| let (mut exec, id, peers, _stream) = setup_connected_peer_test(); |
| |
| let (remote, channel) = Channel::create(); |
| assert!(exec.run_singlethreaded(peers.connected(id, channel, None)).is_ok()); |
| run_to_stalled(&mut exec); |
| |
| // Disconnect the signaling channel, peer should be gone. |
| drop(remote); |
| |
| run_to_stalled(&mut exec); |
| |
| assert!(peers.get(&id).is_none()); |
| |
| // Connect another peer with the same ID |
| let (_remote, channel) = Channel::create(); |
| |
| assert!(exec.run_singlethreaded(peers.connected(id, channel, None)).is_ok()); |
| run_to_stalled(&mut exec); |
| |
| // Should be connected. |
| assert!(peers.get(&id).is_some()); |
| } |
| } |
