fuchsia_bluetooth/types/
channel.rs

1// Copyright 2026 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::endpoints::{ClientEnd, Proxy};
6use fidl_fuchsia_bluetooth as fidl_bt;
7use fidl_fuchsia_bluetooth_bredr as bredr;
8use fuchsia_sync::Mutex;
9use futures::sink::Sink;
10use futures::stream::{FusedStream, Stream};
11use futures::{Future, StreamExt};
12use log::warn;
13use std::fmt;
14use std::pin::Pin;
15use std::sync::Arc;
16use std::task::{Context, Poll};
17
18use crate::error::Error;
19
20pub mod fidl_client;
21pub mod socket;
22
23use fidl_client::FidlClientConnection;
24use socket::SocketConnection;
25
26/// The Channel mode in use for a L2CAP channel.
27#[derive(PartialEq, Debug, Clone)]
28pub enum ChannelMode {
29    Basic,
30    EnhancedRetransmissionMode,
31    LeCreditBasedFlowControl,
32    EnhancedCreditBasedFlowControl,
33}
34
35impl fmt::Display for ChannelMode {
36    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
37        match self {
38            ChannelMode::Basic => write!(f, "Basic"),
39            ChannelMode::EnhancedRetransmissionMode => write!(f, "ERTM"),
40            ChannelMode::LeCreditBasedFlowControl => write!(f, "LE_Credit"),
41            ChannelMode::EnhancedCreditBasedFlowControl => write!(f, "Credit"),
42        }
43    }
44}
45
46pub enum A2dpDirection {
47    Normal,
48    Source,
49    Sink,
50}
51
52impl From<A2dpDirection> for bredr::A2dpDirectionPriority {
53    fn from(pri: A2dpDirection) -> Self {
54        match pri {
55            A2dpDirection::Normal => bredr::A2dpDirectionPriority::Normal,
56            A2dpDirection::Source => bredr::A2dpDirectionPriority::Source,
57            A2dpDirection::Sink => bredr::A2dpDirectionPriority::Sink,
58        }
59    }
60}
61
62impl TryFrom<fidl_bt::ChannelMode> for ChannelMode {
63    type Error = Error;
64    fn try_from(fidl: fidl_bt::ChannelMode) -> Result<Self, Error> {
65        match fidl {
66            fidl_bt::ChannelMode::Basic => Ok(ChannelMode::Basic),
67            fidl_bt::ChannelMode::EnhancedRetransmission => {
68                Ok(ChannelMode::EnhancedRetransmissionMode)
69            }
70            fidl_bt::ChannelMode::LeCreditBasedFlowControl => {
71                Ok(ChannelMode::LeCreditBasedFlowControl)
72            }
73            fidl_bt::ChannelMode::EnhancedCreditBasedFlowControl => {
74                Ok(ChannelMode::EnhancedCreditBasedFlowControl)
75            }
76            x => Err(Error::FailedConversion(format!("Unsupported channel mode type: {x:?}"))),
77        }
78    }
79}
80
81impl From<ChannelMode> for fidl_bt::ChannelMode {
82    fn from(x: ChannelMode) -> Self {
83        match x {
84            ChannelMode::Basic => fidl_bt::ChannelMode::Basic,
85            ChannelMode::EnhancedRetransmissionMode => fidl_bt::ChannelMode::EnhancedRetransmission,
86            ChannelMode::LeCreditBasedFlowControl => fidl_bt::ChannelMode::LeCreditBasedFlowControl,
87            ChannelMode::EnhancedCreditBasedFlowControl => {
88                fidl_bt::ChannelMode::EnhancedCreditBasedFlowControl
89            }
90        }
91    }
92}
93
94#[derive(PartialEq, Debug)]
95pub enum ConnectionBackendType {
96    Socket,
97    FidlClient,
98    FidlServer,
99}
100
101/// A trait representing a Bluetooth data connection.
102/// Concrete implementations handle the specific transport mechanism (e.g., socket or FIDL protocol)
103/// while fulfilling the `Sink` and `Stream` contracts for data transfer.
104pub trait Connection:
105    Stream<Item = Result<Vec<u8>, zx::Status>>
106    + Sink<Vec<u8>, Error = zx::Status>
107    + Send
108    + Sync
109    + std::fmt::Debug
110    + Unpin
111{
112    /// Returns a future that resolves when the connection is closed.
113    fn closed<'a>(&'a self) -> Pin<Box<dyn Future<Output = Result<(), zx::Status>> + 'a>>;
114
115    /// Returns the type of the connection backend.
116    fn connection_type(&self) -> ConnectionBackendType;
117
118    /// Writes data to the connection. This is a non-blocking fast path.
119    /// Returns `SHOULD_WAIT` if the buffer is full.
120    fn write(&self, bytes: &[u8]) -> Result<usize, zx::Status>;
121
122    /// Returns true if the connection is currently closed.
123    fn is_closed(&self) -> bool;
124
125    /// Consumes the connection and returns a partially filled FIDL channel
126    /// containing the transport (e.g., socket handle) if applicable.
127    fn into_fidl_channel(self: Box<Self>) -> Result<bredr::Channel, zx::Status>;
128}
129
130/// A wrapper for Bluetooth channel. Profiles interact with this struct.
131#[derive(Debug)]
132pub struct Channel {
133    pub(crate) connection: Box<dyn Connection>,
134    mode: ChannelMode,
135    max_tx_size: usize,
136    flush_timeout: Arc<Mutex<Option<zx::MonotonicDuration>>>,
137    audio_direction_ext: Option<bredr::AudioDirectionExtProxy>,
138    l2cap_parameters_ext: Option<bredr::L2capParametersExtProxy>,
139    audio_offload_ext: Option<bredr::AudioOffloadExtProxy>,
140    terminated: bool,
141}
142
143impl Channel {
144    pub const DEFAULT_MAX_TX: usize = 672;
145
146    pub fn from_socket(socket: zx::Socket, max_tx_size: usize) -> Result<Self, zx::Status> {
147        let connection = Box::new(SocketConnection::new(socket));
148        Ok(Channel {
149            connection,
150            mode: ChannelMode::Basic,
151            max_tx_size,
152            flush_timeout: Arc::new(Mutex::new(None)),
153            audio_direction_ext: None,
154            l2cap_parameters_ext: None,
155            audio_offload_ext: None,
156            terminated: false,
157        })
158    }
159
160    pub fn from_fidl_client(proxy: fidl_bt::ChannelProxy, max_tx_size: usize) -> Self {
161        let connection = Box::new(FidlClientConnection::new(proxy, max_tx_size));
162        Channel {
163            connection,
164            mode: ChannelMode::Basic,
165            max_tx_size,
166            flush_timeout: Arc::new(Mutex::new(None)),
167            audio_direction_ext: None,
168            l2cap_parameters_ext: None,
169            audio_offload_ext: None,
170            terminated: false,
171        }
172    }
173
174    pub fn from_socket_infallible(socket: zx::Socket, max_tx_size: usize) -> Self {
175        Self::from_socket(socket, max_tx_size).unwrap()
176    }
177
178    pub fn create() -> (Self, Self) {
179        Self::create_with_max_tx(Self::DEFAULT_MAX_TX)
180    }
181
182    pub fn create_with_max_tx(max_tx_size: usize) -> (Self, Self) {
183        let (remote, local) = zx::Socket::create_datagram();
184        (
185            Channel::from_socket(remote, max_tx_size).unwrap(),
186            Channel::from_socket(local, max_tx_size).unwrap(),
187        )
188    }
189
190    pub fn max_tx_size(&self) -> usize {
191        self.max_tx_size
192    }
193
194    pub fn channel_mode(&self) -> &ChannelMode {
195        &self.mode
196    }
197
198    pub fn flush_timeout(&self) -> Option<zx::MonotonicDuration> {
199        self.flush_timeout.lock().clone()
200    }
201
202    pub fn closed<'a>(&'a self) -> impl Future<Output = Result<(), zx::Status>> + 'a {
203        self.connection.closed()
204    }
205
206    pub fn is_closed(&self) -> bool {
207        self.connection.is_closed()
208    }
209
210    pub fn write(&self, bytes: &[u8]) -> Result<usize, zx::Status> {
211        self.connection.write(bytes)
212    }
213
214    pub fn set_audio_priority(
215        &self,
216        dir: A2dpDirection,
217    ) -> impl Future<Output = Result<(), Error>> + use<> {
218        let proxy = self.audio_direction_ext.clone();
219        async move {
220            match proxy {
221                None => return Err(Error::profile("audio priority not supported")),
222                Some(proxy) => proxy
223                    .set_priority(dir.into())
224                    .await?
225                    .map_err(|e| Error::profile(format!("setting priority failed: {e:?}"))),
226            }
227        }
228    }
229
230    pub fn set_flush_timeout(
231        &self,
232        duration: Option<zx::MonotonicDuration>,
233    ) -> impl Future<Output = Result<Option<zx::MonotonicDuration>, Error>> + use<> {
234        let flush_timeout = self.flush_timeout.clone();
235        let current = self.flush_timeout.lock().clone();
236        let proxy = self.l2cap_parameters_ext.clone();
237        async move {
238            match (current, duration) {
239                (None, None) => return Ok(None),
240                (Some(old), Some(new)) if (old - new).into_millis().abs() < 2 => {
241                    return Ok(current);
242                }
243                _ => {}
244            };
245            let proxy =
246                proxy.ok_or_else(|| Error::profile("l2cap parameter changing not supported"))?;
247            let parameters = fidl_bt::ChannelParameters {
248                flush_timeout: duration.clone().map(zx::MonotonicDuration::into_nanos),
249                ..Default::default()
250            };
251            let new_params = proxy.request_parameters(&parameters).await?;
252            let new_timeout = new_params.flush_timeout.map(zx::MonotonicDuration::from_nanos);
253            *(flush_timeout.lock()) = new_timeout.clone();
254            Ok(new_timeout)
255        }
256    }
257
258    pub fn audio_offload(&self) -> Option<bredr::AudioOffloadExtProxy> {
259        self.audio_offload_ext.clone()
260    }
261}
262
263impl Stream for Channel {
264    type Item = Result<Vec<u8>, zx::Status>;
265
266    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
267        let this = self.get_mut();
268        if this.terminated {
269            warn!("Stream was polled after termination");
270            return Poll::Ready(None);
271        }
272        let res = this.connection.poll_next_unpin(cx);
273        if let Poll::Ready(None) = res {
274            this.terminated = true;
275        }
276        res
277    }
278}
279
280impl FusedStream for Channel {
281    fn is_terminated(&self) -> bool {
282        self.terminated
283    }
284}
285
286impl Sink<Vec<u8>> for Channel {
287    type Error = zx::Status;
288
289    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
290        Pin::new(&mut *self.get_mut().connection).poll_ready(cx)
291    }
292
293    fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
294        Pin::new(&mut *self.get_mut().connection).start_send(item)
295    }
296
297    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
298        Pin::new(&mut *self.get_mut().connection).poll_flush(cx)
299    }
300
301    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
302        Pin::new(&mut *self.get_mut().connection).poll_close(cx)
303    }
304}
305
306impl TryFrom<Channel> for bredr::Channel {
307    type Error = Error;
308
309    fn try_from(channel: Channel) -> Result<Self, Self::Error> {
310        let mut fidl_channel = channel
311            .connection
312            .into_fidl_channel()
313            .map_err(|e| Error::profile(format!("Failed to convert to FIDL channel: {e:?}")))?;
314
315        fidl_channel.channel_mode = Some(channel.mode.into());
316        fidl_channel.max_tx_sdu_size = Some(channel.max_tx_size as u16);
317
318        let flush_timeout = channel.flush_timeout.lock().clone();
319        fidl_channel.flush_timeout = flush_timeout.map(zx::MonotonicDuration::into_nanos);
320
321        fidl_channel.ext_direction = channel
322            .audio_direction_ext
323            .map(|proxy| {
324                let chan = proxy.into_channel()?;
325                Ok(ClientEnd::new(chan.into()))
326            })
327            .transpose()
328            .map_err(|_: bredr::AudioDirectionExtProxy| {
329                Error::profile("AudioDirection proxy in use")
330            })?;
331
332        fidl_channel.ext_l2cap = channel
333            .l2cap_parameters_ext
334            .map(|proxy| {
335                let chan = proxy.into_channel()?;
336                Ok(ClientEnd::new(chan.into()))
337            })
338            .transpose()
339            .map_err(|_: bredr::L2capParametersExtProxy| {
340                Error::profile("l2cap parameters proxy in use")
341            })?;
342
343        fidl_channel.ext_audio_offload = channel
344            .audio_offload_ext
345            .map(|proxy| {
346                let chan = proxy.into_channel()?;
347                Ok(ClientEnd::new(chan.into()))
348            })
349            .transpose()
350            .map_err(|_: bredr::AudioOffloadExtProxy| {
351                Error::profile("audio offload proxy in use")
352            })?;
353
354        Ok(fidl_channel)
355    }
356}
357
358impl TryFrom<fidl_fuchsia_bluetooth_bredr::Channel> for Channel {
359    type Error = zx::Status;
360
361    fn try_from(fidl: bredr::Channel) -> Result<Self, Self::Error> {
362        let mode = match fidl.channel_mode.unwrap_or(fidl_bt::ChannelMode::Basic).try_into() {
363            Err(e) => {
364                warn!("Unsupported channel mode type: {e:?}");
365                return Err(zx::Status::INTERNAL);
366            }
367            Ok(c) => c,
368        };
369
370        let max_tx_size = fidl.max_tx_sdu_size.ok_or(zx::Status::INVALID_ARGS)? as usize;
371
372        let connection: Box<dyn Connection> = if let Some(conn) = fidl.connection {
373            let proxy = conn.into_proxy();
374            Box::new(FidlClientConnection::new(proxy, max_tx_size)) as Box<dyn Connection>
375        } else if let Some(socket) = fidl.socket {
376            Box::new(SocketConnection::new(socket)) as Box<dyn Connection>
377        } else {
378            return Err(zx::Status::INVALID_ARGS);
379        };
380
381        Ok(Self {
382            connection,
383            mode,
384            max_tx_size,
385            flush_timeout: Arc::new(Mutex::new(
386                fidl.flush_timeout.map(zx::MonotonicDuration::from_nanos),
387            )),
388            audio_direction_ext: fidl.ext_direction.map(|e| e.into_proxy()),
389            l2cap_parameters_ext: fidl.ext_l2cap.map(|e| e.into_proxy()),
390            audio_offload_ext: fidl.ext_audio_offload.map(|c| c.into_proxy()),
391            terminated: false,
392        })
393    }
394}
395
396#[cfg(test)]
397mod tests {
398    use super::*;
399    use fidl::endpoints::create_request_stream;
400    use fidl_fuchsia_bluetooth as fidl_bt;
401    use fidl_fuchsia_bluetooth_bredr as bredr;
402    use fuchsia_async as fasync;
403    use futures::StreamExt;
404    use std::pin::pin;
405
406    fn build_socket_bredr_channel() -> (bredr::Channel, zx::Socket) {
407        let (remote, local) = zx::Socket::create_datagram();
408        (
409            bredr::Channel {
410                socket: Some(remote),
411                channel_mode: Some(fidl_bt::ChannelMode::Basic),
412                max_tx_sdu_size: Some(1004),
413                ..Default::default()
414            },
415            local,
416        )
417    }
418
419    #[test]
420    fn direction_ext() {
421        let mut exec = fasync::TestExecutor::new();
422
423        let (no_ext, _local) = build_socket_bredr_channel();
424        let channel = Channel::try_from(no_ext).unwrap();
425
426        assert!(
427            exec.run_singlethreaded(channel.set_audio_priority(A2dpDirection::Normal)).is_err()
428        );
429        assert!(exec.run_singlethreaded(channel.set_audio_priority(A2dpDirection::Sink)).is_err());
430
431        let (mut ext, _local) = build_socket_bredr_channel();
432        let (client_end, mut direction_request_stream) =
433            create_request_stream::<bredr::AudioDirectionExtMarker>();
434        ext.ext_direction = Some(client_end);
435
436        let channel = Channel::try_from(ext).unwrap();
437
438        let audio_direction_fut = channel.set_audio_priority(A2dpDirection::Normal);
439        let mut audio_direction_fut = pin!(audio_direction_fut);
440
441        assert!(exec.run_until_stalled(&mut audio_direction_fut).is_pending());
442
443        match exec.run_until_stalled(&mut direction_request_stream.next()) {
444            Poll::Ready(Some(Ok(bredr::AudioDirectionExtRequest::SetPriority {
445                priority,
446                responder,
447            }))) => {
448                assert_eq!(bredr::A2dpDirectionPriority::Normal, priority);
449                responder.send(Ok(())).expect("response to send cleanly");
450            }
451            x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
452        };
453
454        match exec.run_until_stalled(&mut audio_direction_fut) {
455            Poll::Ready(Ok(())) => {}
456            _x => panic!("Expected ok result from audio direction response"),
457        };
458
459        let audio_direction_fut = channel.set_audio_priority(A2dpDirection::Sink);
460        let mut audio_direction_fut = pin!(audio_direction_fut);
461
462        assert!(exec.run_until_stalled(&mut audio_direction_fut).is_pending());
463
464        match exec.run_until_stalled(&mut direction_request_stream.next()) {
465            Poll::Ready(Some(Ok(bredr::AudioDirectionExtRequest::SetPriority {
466                priority,
467                responder,
468            }))) => {
469                assert_eq!(bredr::A2dpDirectionPriority::Sink, priority);
470                responder
471                    .send(Err(fidl_fuchsia_bluetooth::ErrorCode::Failed))
472                    .expect("response to send cleanly");
473            }
474            x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
475        };
476
477        match exec.run_until_stalled(&mut audio_direction_fut) {
478            Poll::Ready(Err(_)) => {}
479            _x => panic!("Expected error result from audio direction response"),
480        };
481    }
482
483    #[test]
484    fn flush_timeout() {
485        let mut exec = fasync::TestExecutor::new();
486
487        let (mut no_ext, _local) = build_socket_bredr_channel();
488        no_ext.flush_timeout = Some(50_000_000); // 50 milliseconds
489        let channel = Channel::try_from(no_ext).unwrap();
490
491        assert_eq!(Some(zx::MonotonicDuration::from_millis(50)), channel.flush_timeout());
492
493        // Within 2 milliseconds, doesn't change.
494        let res = exec.run_singlethreaded(
495            channel.set_flush_timeout(Some(zx::MonotonicDuration::from_millis(49))),
496        );
497        assert_eq!(Some(zx::MonotonicDuration::from_millis(50)), res.expect("shouldn't error"));
498        let res = exec.run_singlethreaded(
499            channel.set_flush_timeout(Some(zx::MonotonicDuration::from_millis(51))),
500        );
501        assert_eq!(Some(zx::MonotonicDuration::from_millis(50)), res.expect("shouldn't error"));
502
503        assert!(
504            exec.run_singlethreaded(
505                channel.set_flush_timeout(Some(zx::MonotonicDuration::from_millis(200)))
506            )
507            .is_err()
508        );
509        assert!(exec.run_singlethreaded(channel.set_flush_timeout(None)).is_err());
510
511        let (mut ext, _local) = build_socket_bredr_channel();
512        let (client_end, mut l2cap_request_stream) =
513            create_request_stream::<bredr::L2capParametersExtMarker>();
514        ext.ext_l2cap = Some(client_end);
515
516        let channel = Channel::try_from(ext).unwrap();
517
518        {
519            let flush_timeout_fut = channel.set_flush_timeout(None);
520            let mut flush_timeout_fut = pin!(flush_timeout_fut);
521
522            // Requesting no change returns right away with no change.
523            match exec.run_until_stalled(&mut flush_timeout_fut) {
524                Poll::Ready(Ok(None)) => {}
525                x => panic!("Expected no flush timeout to not stall, got {:?}", x),
526            }
527        }
528
529        let req_duration = zx::MonotonicDuration::from_millis(42);
530
531        {
532            let flush_timeout_fut = channel.set_flush_timeout(Some(req_duration));
533            let mut flush_timeout_fut = pin!(flush_timeout_fut);
534
535            assert!(exec.run_until_stalled(&mut flush_timeout_fut).is_pending());
536
537            match exec.run_until_stalled(&mut l2cap_request_stream.next()) {
538                Poll::Ready(Some(Ok(bredr::L2capParametersExtRequest::RequestParameters {
539                    request,
540                    responder,
541                }))) => {
542                    assert_eq!(Some(req_duration.into_nanos()), request.flush_timeout);
543                    // Send a different response
544                    let params = fidl_bt::ChannelParameters {
545                        flush_timeout: Some(50_000_000), // 50ms
546                        ..Default::default()
547                    };
548                    responder.send(&params).expect("response to send cleanly");
549                }
550                x => panic!("Expected a item to be ready on the request stream, got {:?}", x),
551            };
552
553            match exec.run_until_stalled(&mut flush_timeout_fut) {
554                Poll::Ready(Ok(Some(duration))) => {
555                    assert_eq!(zx::MonotonicDuration::from_millis(50), duration)
556                }
557                x => panic!("Expected ready result from params response, got {:?}", x),
558            };
559        }
560
561        // Channel should have recorded the new flush timeout.
562        assert_eq!(Some(zx::MonotonicDuration::from_millis(50)), channel.flush_timeout());
563    }
564
565    #[test]
566    fn audio_offload() {
567        let _exec = fasync::TestExecutor::new();
568
569        let (no_ext, _local) = build_socket_bredr_channel();
570        let channel = Channel::try_from(no_ext).unwrap();
571
572        assert!(channel.audio_offload().is_none());
573
574        let (mut ext, _local) = build_socket_bredr_channel();
575        let (client_end, mut _audio_offload_ext_req_stream) =
576            create_request_stream::<bredr::AudioOffloadExtMarker>();
577        ext.ext_audio_offload = Some(client_end);
578
579        let channel = Channel::try_from(ext).unwrap();
580
581        let offload_ext = channel.audio_offload();
582        assert!(offload_ext.is_some());
583        // We can get the audio offload multiple times without dropping
584        assert!(channel.audio_offload().is_some());
585        // And with dropping
586        drop(offload_ext);
587        assert!(channel.audio_offload().is_some());
588    }
589
590    #[test]
591    fn channel_from_fidl_priority() {
592        let _exec = fasync::TestExecutor::new();
593
594        // Case 1: Both FIDL connection and socket are present.
595        // FIDL connection should be preferred over socket.
596        let (client_end, _server_end) =
597            fidl::endpoints::create_endpoints::<fidl_bt::ChannelMarker>();
598        let (mut fidl_both, _socket_local) = build_socket_bredr_channel();
599        fidl_both.connection = Some(client_end);
600
601        let chan = Channel::try_from(fidl_both).expect("to convert successfully");
602        assert_eq!(chan.connection.connection_type(), ConnectionBackendType::FidlClient);
603
604        // Case 2: Only socket is present.
605        // Should fall back to traditional socket transport.
606        let (socket_only, _socket_local) = build_socket_bredr_channel();
607
608        let chan = Channel::try_from(socket_only).expect("to convert successfully");
609        assert_eq!(chan.connection.connection_type(), ConnectionBackendType::Socket);
610
611        // Case 3: Neither is present.
612        // Should fail to convert as we need at least one transport.
613        let fidl_empty = bredr::Channel {
614            channel_mode: Some(fidl_bt::ChannelMode::Basic),
615            max_tx_sdu_size: Some(1004),
616            ..Default::default()
617        };
618        assert!(Channel::try_from(fidl_empty).is_err());
619    }
620}
fuchsia_bluetooth/types/channel.rs

fuchsia_bluetooth/types/
channel.rs
