test_rfcomm_client/
lib.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::{Error, format_err};
6use bt_rfcomm::ServerChannel;
7use bt_rfcomm::profile::server_channel_from_protocol;
8use derivative::Derivative;
9use fidl_fuchsia_bluetooth as fidl_bt;
10use fidl_fuchsia_bluetooth_bredr as bredr;
11use fidl_fuchsia_bluetooth_rfcomm_test as rfcomm;
12use fuchsia_async as fasync;
13use fuchsia_bluetooth::profile::ProtocolDescriptor;
14use fuchsia_bluetooth::types::{Channel, PeerId, Uuid};
15use fuchsia_sync::Mutex;
16use futures::channel::mpsc;
17use futures::{SinkExt, StreamExt, select};
18use log::{info, warn};
19use profile_client::{ProfileClient, ProfileEvent};
20use std::cell::Cell;
21use std::collections::HashMap;
22use std::sync::Arc;
23
24/// The default buffer size for the mpsc channels used to relay user data packets to be sent to the
25/// remote peer.
26/// This value is arbitrarily chosen and should be enough to queue multiple buffers to be sent.
27const USER_DATA_BUFFER_SIZE: usize = 50;
28
29/// Valid SPP Service Definition - see SPP v1.2 Table 6.1.
30fn spp_service_definition() -> bredr::ServiceDefinition {
31    bredr::ServiceDefinition {
32        service_class_uuids: Some(vec![
33            Uuid::new16(bredr::ServiceClassProfileIdentifier::SerialPort.into_primitive()).into(),
34        ]),
35        protocol_descriptor_list: Some(vec![
36            bredr::ProtocolDescriptor {
37                protocol: Some(bredr::ProtocolIdentifier::L2Cap),
38                params: Some(vec![]),
39                ..Default::default()
40            },
41            bredr::ProtocolDescriptor {
42                protocol: Some(bredr::ProtocolIdentifier::Rfcomm),
43                params: Some(vec![]),
44                ..Default::default()
45            },
46        ]),
47        profile_descriptors: Some(vec![bredr::ProfileDescriptor {
48            profile_id: Some(bredr::ServiceClassProfileIdentifier::SerialPort),
49            major_version: Some(1),
50            minor_version: Some(2),
51            ..Default::default()
52        }]),
53        ..Default::default()
54    }
55}
56
57/// Manages the set of active RFCOMM channels connected to a single remote peer.
58#[derive(Debug)]
59pub struct RfcommSession {
60    /// Unique id assigned to the remote peer.
61    _id: PeerId,
62    /// The set of active RFCOMM channels.
63    active_channels: HashMap<ServerChannel, mpsc::Sender<Vec<u8>>>,
64}
65
66impl RfcommSession {
67    fn new(id: PeerId) -> Self {
68        Self { _id: id, active_channels: HashMap::new() }
69    }
70
71    fn is_active(&self, server_channel: &ServerChannel) -> bool {
72        self.active_channels.get(server_channel).is_some_and(|s| !s.is_closed())
73    }
74
75    fn close_rfcomm_channel(&mut self, server_channel: &ServerChannel) -> bool {
76        self.active_channels.remove(server_channel).is_some()
77    }
78
79    fn new_rfcomm_channel(&mut self, server_channel: ServerChannel, channel: Channel) {
80        if self.is_active(&server_channel) {
81            info!("Overwriting existing RFCOMM channel: {:?}", server_channel);
82        }
83
84        let (sender, receiver) = mpsc::channel(USER_DATA_BUFFER_SIZE);
85        fasync::Task::spawn(Self::rfcomm_channel_task(server_channel, channel, receiver)).detach();
86        let _ = self.active_channels.insert(server_channel, sender);
87    }
88
89    /// Processes data received from the remote peer over the provided RFCOMM `channel`.
90    /// Processes data in the `write_requests` queue to be sent to the remote peer.
91    async fn rfcomm_channel_task(
92        server_channel: ServerChannel,
93        mut channel: Channel,
94        mut write_requests: mpsc::Receiver<Vec<u8>>,
95    ) {
96        info!("Starting processing task for RFCOMM channel: {:?}", server_channel);
97        loop {
98            select! {
99                // The `fuse()` call is in the loop because `channel` is both borrowed as a stream
100                // and used to send data. It is safe because once `channel` is closed, the loop will
101                // break and `channel.next()` will never be polled thereafter.
102                bytes_from_peer = channel.next() => {
103                    let user_data = match bytes_from_peer {
104                        Some(Ok(bytes)) => bytes,
105                        Some(Err(e)) => {
106                            info!("Error receiving data: {:?}", e);
107                            continue;
108                        }
109                        None => {
110                            // RFCOMM channel closed by the peer.
111                            info!("Peer closed RFCOMM channel {:?}", server_channel);
112                            break;
113                        }
114                    };
115                    info!("{:?}: Received user data from peer: {:?}", server_channel, user_data);
116                }
117                bytes_to_peer = write_requests.next() => {
118                    match bytes_to_peer {
119                        Some(bytes) => {
120                            match channel.send(bytes).await {
121                                Ok(()) => info!("Sent user data over RFCOMM channel ({:?}).", server_channel),
122                                Err(e) => info!("Couldn't send user data for channel ({:?}): {:?}", server_channel, e),
123                            }
124                        }
125                        None => break, // RFCOMM channel closed by us.
126                    }
127                }
128                complete => break,
129            }
130        }
131        info!("RFCOMM channel ({:?}) task ended", server_channel);
132    }
133
134    /// Sends the `user_data` buf to the peer that provides the service identified by the
135    /// `server_channel`. Returns the result of the send operation.
136    fn send_user_data(
137        &mut self,
138        server_channel: ServerChannel,
139        user_data: Vec<u8>,
140    ) -> Result<(), Error> {
141        if let Some(sender) = self.active_channels.get_mut(&server_channel) {
142            sender.try_send(user_data).map_err(|e| format_err!("{:?}", e))
143        } else {
144            Err(format_err!("No registered server channel"))
145        }
146    }
147}
148
149#[derive(Derivative, Default)]
150#[derivative(Debug)]
151pub struct RfcommState {
152    /// A task representing the RFCOMM service advertisement and search.
153    #[derivative(Debug = "ignore")]
154    service: Option<fasync::Task<()>>,
155    /// The set of active RFCOMM Sessions with remote peers.
156    active_sessions: HashMap<PeerId, RfcommSession>,
157}
158
159impl RfcommState {
160    fn new() -> Self {
161        Self { service: None, active_sessions: HashMap::new() }
162    }
163
164    fn get_active_session(&mut self, id: &PeerId) -> Option<&mut RfcommSession> {
165        match self.active_sessions.get_mut(id) {
166            None => {
167                info!("No active RFCOMM session with peer {}", id);
168                None
169            }
170            session => session,
171        }
172    }
173
174    fn clear_services(&mut self) {
175        if let Some(old_task) = self.service.take() {
176            info!("Clearing SPP service advertisement/search");
177            let _ = old_task.abort();
178        }
179        self.active_sessions.clear();
180    }
181
182    fn new_rfcomm_channel(&mut self, id: PeerId, server_channel: ServerChannel, channel: Channel) {
183        let _ = self
184            .active_sessions
185            .entry(id)
186            .or_insert_with(|| RfcommSession::new(id))
187            .new_rfcomm_channel(server_channel, channel);
188    }
189}
190
191#[derive(Derivative, Default)]
192#[derivative(Debug)]
193pub struct RfcommManager {
194    #[derivative(Debug = "ignore")]
195    profile: Cell<Option<bredr::ProfileProxy>>,
196    #[derivative(Debug = "ignore")]
197    rfcomm: Cell<Option<rfcomm::RfcommTestProxy>>,
198    inner: Arc<Mutex<RfcommState>>,
199}
200
201impl Clone for RfcommManager {
202    fn clone(&self) -> Self {
203        let profile = self.profile.take();
204        if let Some(p) = profile.as_ref() {
205            self.profile.set(Some(p.clone()));
206        }
207        let rfcomm = self.rfcomm.take();
208        if let Some(rf) = rfcomm.as_ref() {
209            self.rfcomm.set(Some(rf.clone()));
210        }
211        Self { profile: Cell::new(profile), rfcomm: Cell::new(rfcomm), inner: self.inner.clone() }
212    }
213}
214
215impl RfcommManager {
216    pub fn new() -> Result<Self, Error> {
217        Ok(Self::default())
218    }
219
220    pub fn from_proxy(profile: bredr::ProfileProxy, rfcomm: rfcomm::RfcommTestProxy) -> Self {
221        Self {
222            profile: Cell::new(Some(profile)),
223            rfcomm: Cell::new(Some(rfcomm)),
224            inner: Arc::new(Mutex::new(RfcommState::new())),
225        }
226    }
227
228    pub fn clear_services(&self) {
229        self.inner.lock().clear_services();
230    }
231
232    fn get_profile_proxy(&self) -> Result<bredr::ProfileProxy, Error> {
233        let proxy = match self.profile.take() {
234            Some(proxy) => proxy,
235            None => fuchsia_component::client::connect_to_protocol::<bredr::ProfileMarker>()?,
236        };
237        self.profile.set(Some(proxy.clone()));
238        Ok(proxy)
239    }
240
241    fn get_rfcomm_test_proxy(&self) -> Result<rfcomm::RfcommTestProxy, Error> {
242        let proxy = match self.rfcomm.take() {
243            Some(proxy) => proxy,
244            None => fuchsia_component::client::connect_to_protocol::<rfcomm::RfcommTestMarker>()?,
245        };
246        self.rfcomm.set(Some(proxy.clone()));
247        Ok(proxy)
248    }
249
250    /// Advertises an SPP service and searches for other compatible SPP clients. Overwrites any
251    /// existing service advertisement & search.
252    pub fn advertise(&self) -> Result<(), Error> {
253        // Existing service must be unregistered before we can advertise again - this is to prevent
254        // clashes in `bredr.Profile` server.
255        self.clear_services();
256
257        let profile_proxy = self.get_profile_proxy()?;
258        let inner_clone = self.inner.clone();
259        let mut inner = self.inner.lock();
260
261        // Add an SPP advertisement & search.
262        let spp_service = vec![spp_service_definition()];
263        let mut client = ProfileClient::advertise(
264            profile_proxy,
265            spp_service,
266            fidl_bt::ChannelParameters::default(),
267        )?;
268        let _ = client.add_search(bredr::ServiceClassProfileIdentifier::SerialPort, None)?;
269        let service_task = fasync::Task::spawn(async move {
270            let result = Self::handle_profile_events(client, inner_clone).await;
271            info!("Profile event handler ended: {:?}", result);
272        });
273        inner.service = Some(service_task);
274        info!("Advertising and searching for SPP services");
275        Ok(())
276    }
277
278    /// Processes events from the `bredr.Profile` `client`.
279    async fn handle_profile_events(
280        mut client: ProfileClient,
281        state: Arc<Mutex<RfcommState>>,
282    ) -> Result<(), Error> {
283        while let Some(request) = client.next().await {
284            match request {
285                Ok(ProfileEvent::PeerConnected { id, protocol, channel, .. }) => {
286                    // Received an incoming connection request for our advertised service.
287                    let protocol = protocol
288                        .iter()
289                        .map(|p| ProtocolDescriptor::try_from(p))
290                        .collect::<Result<Vec<_>, _>>()?;
291                    let server_channel = server_channel_from_protocol(&protocol)
292                        .ok_or_else(|| format_err!("Not RFCOMM protocol"))?;
293
294                    // Spawn a processing task to handle read & writes over this RFCOMM channel.
295                    state.lock().new_rfcomm_channel(id, server_channel, channel);
296                    info!("Peer {} established RFCOMM Channel ({:?}) ", id, server_channel);
297                }
298                Ok(ProfileEvent::SearchResult { id, protocol, .. }) => {
299                    // Discovered a remote peer's service.
300                    let protocol = protocol
301                        .expect("Protocol should exist")
302                        .iter()
303                        .map(|p| ProtocolDescriptor::try_from(p))
304                        .collect::<Result<Vec<_>, _>>()?;
305                    let server_channel = server_channel_from_protocol(&protocol)
306                        .ok_or_else(|| format_err!("Not RFCOMM protocol"))?;
307                    info!("Found SPP service for {} with server channel: {:?}", id, server_channel);
308                }
309                Err(e) => warn!("Error in ProfileClient results: {:?}", e),
310            }
311        }
312        Ok(())
313    }
314
315    /// Terminates the RFCOMM session with the remote peer `id`.
316    pub fn close_session(&self, id: PeerId) -> Result<(), Error> {
317        // Send the disconnect request via the `RfcommTest` API and clean up local state.
318        let _ = self
319            .get_rfcomm_test_proxy()?
320            .disconnect(&id.into())
321            .map_err::<fidl::Error, _>(Into::into)?;
322
323        let mut inner = self.inner.lock();
324        if let Some(session) = inner.active_sessions.remove(&id) {
325            drop(session);
326        }
327        Ok(())
328    }
329
330    /// Closes the RFCOMM channel with the remote peer.
331    pub fn close_rfcomm_channel(
332        &self,
333        id: PeerId,
334        server_channel: ServerChannel,
335    ) -> Result<(), Error> {
336        let mut inner = self.inner.lock();
337        if let Some(session) = inner.get_active_session(&id) {
338            let _ = session.close_rfcomm_channel(&server_channel);
339            Ok(())
340        } else {
341            Err(format_err!("No RFCOMM session with peer: {:?}", id))
342        }
343    }
344
345    /// Makes an outgoing RFCOMM channel to the remote peer.
346    pub async fn outgoing_rfcomm_channel(
347        &self,
348        id: PeerId,
349        server_channel: ServerChannel,
350    ) -> Result<(), Error> {
351        let channel = self
352            .get_profile_proxy()?
353            .connect(
354                &id.into(),
355                &bredr::ConnectParameters::Rfcomm(bredr::RfcommParameters {
356                    channel: Some(server_channel.into()),
357                    ..Default::default()
358                }),
359            )
360            .await?
361            .map_err(|e| format_err!("{:?}", e))?;
362        let channel = Channel::try_from(channel).expect("valid channel");
363
364        self.inner.lock().new_rfcomm_channel(id, server_channel, channel);
365        Ok(())
366    }
367
368    /// Send a Remote Line Status update for the RFCOMM `server_channel` with peer `id`. Returns
369    /// Error if there is no such established RFCOMM channel with the peer.
370    pub fn send_rls(&self, id: PeerId, server_channel: ServerChannel) -> Result<(), Error> {
371        let rfcomm_test_proxy = self.get_rfcomm_test_proxy()?;
372        let mut inner = self.inner.lock();
373        if inner.get_active_session(&id).is_some() {
374            // Send a fixed Framing error status.
375            let status = rfcomm::Status::FramingError;
376            let _ = rfcomm_test_proxy
377                .remote_line_status(&id.into(), server_channel.into(), status)
378                .map_err::<fidl::Error, _>(Into::into)?;
379            Ok(())
380        } else {
381            Err(format_err!("No RFCOMM session with peer: {:?}", id))
382        }
383    }
384
385    /// Attempts to send user `data` to the remote peer `id`. Returns Error if there is no such
386    /// established RFCOMM channel with the peer.
387    pub fn send_user_data(
388        &self,
389        id: PeerId,
390        server_channel: ServerChannel,
391        data: Vec<u8>,
392    ) -> Result<(), Error> {
393        let mut inner = self.inner.lock();
394        if let Some(session) = inner.get_active_session(&id) {
395            session.send_user_data(server_channel, data)
396        } else {
397            Err(format_err!("No RFCOMM session with peer: {:?}", id))
398        }
399    }
400}
401
402#[cfg(test)]
403mod tests {
404    use super::*;
405    use assert_matches::assert_matches;
406    use async_utils::PollExt;
407    use bt_channel_test_support::{Transport, create_test_channels};
408    use bt_rfcomm::profile::build_rfcomm_protocol;
409    use fidl::endpoints::Proxy;
410    use fidl_fuchsia_bluetooth::ErrorCode;
411    use fidl_fuchsia_bluetooth_bredr::{ProfileMarker, ProfileRequestStream};
412    use fidl_fuchsia_bluetooth_rfcomm_test::{RfcommTestMarker, RfcommTestRequestStream};
413    use fixture::fixture;
414    use test_case::test_case;
415
416    type TestFixture = (RfcommManager, ProfileRequestStream, RfcommTestRequestStream);
417
418    async fn setup_rfcomm_mgr<F, Fut>(_name: &str, test: F)
419    where
420        F: FnOnce(TestFixture) -> Fut,
421        Fut: futures::Future<Output = ()>,
422    {
423        let (profile, profile_server) = fidl::endpoints::create_proxy_and_stream::<ProfileMarker>();
424        let (rfcomm_test, rfcomm_test_server) =
425            fidl::endpoints::create_proxy_and_stream::<RfcommTestMarker>();
426
427        let rfcomm_mgr = RfcommManager::from_proxy(profile, rfcomm_test);
428        test((rfcomm_mgr, profile_server, rfcomm_test_server)).await
429    }
430
431    async fn expect_data(remote: &mut Channel, expected_data: Vec<u8>) {
432        let read_result = remote.next().await.expect("data").expect("okay");
433        assert_eq!(read_result, expected_data);
434    }
435
436    async fn expect_advertisement_and_search(
437        profile: &mut ProfileRequestStream,
438    ) -> (
439        bredr::SearchResultsProxy,
440        (bredr::ConnectionReceiverProxy, bredr::ProfileAdvertiseResponder),
441    ) {
442        let mut search_request = None;
443        let mut advertisement = None;
444        while let Some(req) = profile.next().await {
445            match req {
446                Ok(bredr::ProfileRequest::Advertise { payload, responder, .. }) => {
447                    let connect_proxy = payload.receiver.unwrap().into_proxy();
448                    advertisement = Some((connect_proxy, responder));
449                }
450                Ok(bredr::ProfileRequest::Search { payload, .. }) => {
451                    search_request = Some(payload.results.unwrap().into_proxy())
452                }
453                x => panic!("Expected one Advertise and Search but got: {:?}", x),
454            }
455            if search_request.is_some() && advertisement.is_some() {
456                break;
457            }
458        }
459        (search_request.expect("just set"), advertisement.expect("just set"))
460    }
461
462    #[fixture(setup_rfcomm_mgr)]
463    #[fuchsia::test]
464    async fn initiate_rfcomm_channel_to_peer_is_ok(
465        (rfcomm_mgr, mut profile_server, mut rfcomm_test_server): TestFixture,
466    ) {
467        // Keep the `bredr.Profile` requests alive - one advertisement and search.
468        let _profile_requests = {
469            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
470            expect_advertisement_and_search(&mut profile_server).await
471        };
472
473        // Can establish RFCOMM channel to peer.
474        let remote_id = PeerId(123);
475        let random_channel_number = ServerChannel::try_from(5).unwrap();
476        let mut peer_channel = {
477            let ch_fut =
478                Box::pin(rfcomm_mgr.outgoing_rfcomm_channel(remote_id, random_channel_number));
479
480            let profile_fut = async {
481                match profile_server.next().await {
482                    Some(Ok(bredr::ProfileRequest::Connect { responder, .. })) => {
483                        let (left, right) = Channel::create_socket_pair();
484                        let _ = responder
485                            .send(left.try_into().map_err(|_e| ErrorCode::Failed))
486                            .unwrap();
487                        right
488                    }
489                    x => panic!("Expected connect request, got: {:?}", x),
490                }
491            };
492
493            match futures::future::join(ch_fut, profile_fut).await {
494                (Ok(_), channel) => channel,
495                x => panic!("Expected both futures to complete: {:?}", x),
496            }
497        };
498
499        // Sending data to the peer is ok.
500        let user_data = vec![0x98, 0x97, 0x96, 0x95];
501        {
502            assert_matches!(
503                rfcomm_mgr.send_user_data(remote_id, random_channel_number, user_data.clone()),
504                Ok(_)
505            );
506            expect_data(&mut peer_channel, user_data).await;
507        }
508
509        // Peer sends us data. It should be received gracefully and logged (nothing to test).
510        let buf = vec![0x99, 0x11, 0x44];
511        assert_matches!(peer_channel.send(buf).await, Ok(()));
512
513        // Test client can request to send an RLS update - should be received by RFCOMM Test server.
514        assert_matches!(rfcomm_mgr.send_rls(remote_id, random_channel_number), Ok(_));
515        match rfcomm_test_server.next().await.expect("valid fidl request") {
516            Ok(rfcomm::RfcommTestRequest::RemoteLineStatus { id, channel_number, .. }) => {
517                assert_eq!(id, remote_id.into());
518                assert_eq!(channel_number, u8::from(random_channel_number));
519            }
520            x => panic!("Expected RLS request but got: {:?}", x),
521        }
522    }
523
524    #[fixture(setup_rfcomm_mgr)]
525    #[fuchsia::test]
526    async fn peer_initiating_rfcomm_channel_is_delivered(
527        (rfcomm_mgr, mut profile_server, _rfcomm_test_server): TestFixture,
528    ) {
529        // Keep the `bredr.Profile` requests alive - one advertisement and search.
530        let (_search_proxy, (connect_proxy, _adv_fut)) = {
531            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
532            expect_advertisement_and_search(&mut profile_server).await
533        };
534
535        // Peer connects to us.
536        let remote_id = PeerId(8978);
537        let random_channel_number = ServerChannel::try_from(7).unwrap();
538        let (_peer_channel, local_channel) = Channel::create_socket_pair();
539        let protocol: Vec<bredr::ProtocolDescriptor> =
540            build_rfcomm_protocol(random_channel_number).iter().map(Into::into).collect();
541        assert_matches!(
542            connect_proxy.connected(
543                &remote_id.into(),
544                local_channel.try_into().unwrap(),
545                &protocol,
546            ),
547            Ok(_)
548        );
549    }
550
551    #[fixture(setup_rfcomm_mgr)]
552    #[fuchsia::test]
553    async fn disconnect_session_received_by_rfcomm_test(
554        (rfcomm_mgr, mut profile_server, mut rfcomm_test_server): TestFixture,
555    ) {
556        // Keep the `bredr.Profile` requests alive - one advertisement and search.
557        let _profile_requests = {
558            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
559            expect_advertisement_and_search(&mut profile_server).await
560        };
561
562        // Even though there are no active RFCOMM channels established, a client can still request
563        // to disconnect the session - expect it to be received.
564        let remote = PeerId(834);
565        assert_matches!(rfcomm_mgr.close_session(remote), Ok(_));
566
567        match rfcomm_test_server.next().await.expect("valid fidl request") {
568            Ok(rfcomm::RfcommTestRequest::Disconnect { id, .. }) if id == remote.into() => {}
569            x => panic!("Expected Disconnect request but got: {:?}", x),
570        }
571    }
572
573    #[fixture(setup_rfcomm_mgr)]
574    #[fuchsia::test]
575    async fn rls_update_before_established_channel_is_error(
576        (rfcomm_mgr, mut profile_server, _rfcomm_test_server): TestFixture,
577    ) {
578        // Keep the `bredr.Profile` requests alive - one advertisement and search.
579        let _profile_requests = {
580            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
581            expect_advertisement_and_search(&mut profile_server).await
582        };
583
584        // RLS updates pertain to a specific RFCOMM channel. Expect an error if an RLS request is
585        // sent for a non existent channel.
586        let remote = PeerId(222);
587        let random_channel_number = ServerChannel::try_from(9).unwrap();
588        assert_matches!(rfcomm_mgr.send_rls(remote, random_channel_number), Err(_));
589    }
590
591    #[fixture(setup_rfcomm_mgr)]
592    #[fuchsia::test]
593    async fn clear_services_unregisters_profile_requests(
594        (rfcomm_mgr, mut profile_server, _rfcomm_test_server): TestFixture,
595    ) {
596        // Keep the `bredr.Profile` requests alive - one advertisement and search.
597        let (search_proxy, (connect_proxy, _advertise_fut)) = {
598            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
599            expect_advertisement_and_search(&mut profile_server).await
600        };
601        assert!(!search_proxy.is_closed());
602        assert!(!connect_proxy.is_closed());
603
604        // Clearing services should unregister advertisement and search (transitively closing the
605        // FIDL channels).
606        // Note: Clearing `Profile` services cancels the fasync::Task processing the `bredr.Profile`
607        // requests. Per documentation of fasync::Task, there are no guarantees about the freeing
608        // of resources held by a Task. Therefore, we cannot assume `search_proxy` and
609        // `connect_proxy` will be closed immediately (but we do expect them to be freed eventually)
610        rfcomm_mgr.clear_services();
611
612        // Can register again.
613        let _profile = {
614            assert_matches!(rfcomm_mgr.advertise(), Ok(_));
615            expect_advertisement_and_search(&mut profile_server).await
616        };
617    }
618
619    #[test_case(Transport::Socket ; "socket")]
620    #[test_case(Transport::Fidl ; "fidl")]
621    #[fuchsia::test]
622    async fn rfcomm_session_task(transport: Transport) {
623        let id = PeerId(999);
624        let mut session = RfcommSession::new(id);
625
626        let random_channel_number = ServerChannel::try_from(4).unwrap();
627        let (local, mut remote) = create_test_channels(transport);
628        session.new_rfcomm_channel(random_channel_number, local);
629
630        assert!(session.is_active(&random_channel_number));
631
632        let data = vec![0x00, 0x02, 0x04, 0x06, 0x08, 0x10];
633        let unregistered = ServerChannel::try_from(9).unwrap();
634        // Unregistered channel number is error.
635        assert_matches!(session.send_user_data(unregistered, data.clone()), Err(_));
636        // Sending is OK.
637        assert_matches!(session.send_user_data(random_channel_number, data.clone()), Ok(_));
638
639        // Should be received by remote.
640        expect_data(&mut remote, data).await;
641
642        // Can send multiple buffers.
643        let data1 = vec![0x09];
644        let data2 = vec![0x11];
645        assert_matches!(session.send_user_data(random_channel_number, data1.clone()), Ok(_));
646        assert_matches!(session.send_user_data(random_channel_number, data2.clone()), Ok(_));
647        expect_data(&mut remote, data1).await;
648        expect_data(&mut remote, data2).await;
649
650        // Local wants to close channel - should disconnect.
651        assert!(session.close_rfcomm_channel(&random_channel_number));
652        assert_matches!(remote.closed().await, Ok(_));
653
654        // Trying again is OK - nothing happens.
655        assert!(!session.close_rfcomm_channel(&random_channel_number));
656    }
657
658    #[test_case(Transport::Socket ; "socket")]
659    #[test_case(Transport::Fidl ; "fidl")]
660    #[fuchsia::test]
661    async fn second_channel_overwrites_first_in_rfcomm_session(transport: Transport) {
662        let id = PeerId(78);
663        let mut session = RfcommSession::new(id);
664
665        let random_channel_number = ServerChannel::try_from(10).unwrap();
666        let (local1, remote1) = create_test_channels(transport);
667        session.new_rfcomm_channel(random_channel_number, local1);
668        assert!(session.is_active(&random_channel_number));
669
670        // Can create a new RFCOMM channel, this will overwrite the existing one.
671        let (local2, mut remote2) = create_test_channels(transport);
672        session.new_rfcomm_channel(random_channel_number, local2);
673        assert!(session.is_active(&random_channel_number));
674
675        assert_matches!(remote1.closed().await, Ok(_));
676
677        let data = vec![0x00, 0x02, 0x04, 0x06, 0x08, 0x10];
678        // Sending is OK - should be received by remote.
679        assert_matches!(session.send_user_data(random_channel_number, data.clone()), Ok(_));
680        expect_data(&mut remote2, data).await;
681    }
682
683    #[test_case(Transport::Socket ; "socket")]
684    #[test_case(Transport::Fidl ; "fidl")]
685    #[fuchsia::test]
686    fn closing_sender_closes_rfcomm_channel_task(transport: Transport) {
687        let mut exec = fasync::TestExecutor::new();
688
689        let random_channel_number = ServerChannel::try_from(10).unwrap();
690        let (local, _remote) = create_test_channels(transport);
691        let (_sender, receiver) = mpsc::channel(0);
692
693        let mut channel_task =
694            Box::pin(RfcommSession::rfcomm_channel_task(random_channel_number, local, receiver));
695
696        exec.run_until_stalled(&mut channel_task).expect_pending("sender still active");
697
698        drop(_sender);
699        let _ = exec.run_until_stalled(&mut channel_task).expect("task should complete");
700    }
701
702    #[test_case(Transport::Socket ; "socket")]
703    #[test_case(Transport::Fidl ; "fidl")]
704    #[fuchsia::test]
705    fn closing_channel_closes_rfcomm_channel_task(transport: Transport) {
706        let mut exec = fasync::TestExecutor::new();
707
708        let random_channel_number = ServerChannel::try_from(10).unwrap();
709        let (local, _remote) = create_test_channels(transport);
710        let (_sender, receiver) = mpsc::channel(0);
711
712        let mut channel_task =
713            Box::pin(RfcommSession::rfcomm_channel_task(random_channel_number, local, receiver));
714
715        exec.run_until_stalled(&mut channel_task).expect_pending("sender still active");
716
717        drop(_remote);
718        let _ = exec.run_until_stalled(&mut channel_task).expect("task should complete");
719    }
720}
test_rfcomm_client/lib.rs

test_rfcomm_client/
lib.rs
