fuchsia_bluetooth/types/channel/
socket.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_fuchsia_bluetooth_bredr as bredr;
6use fuchsia_async as fasync;
7use futures::sink::Sink;
8use futures::stream::Stream;
9use futures::{Future, TryFutureExt, ready};
10use log::error;
11use std::collections::VecDeque;
12use std::pin::Pin;
13use std::task::{Context, Poll};
14use zx_status_ext::IoErrorKindExt;
15
16use super::{Connection, ConnectionBackendType};
17
18/// A socket-based implementation of the Bluetooth channel transport.
19#[derive(Debug)]
20pub struct SocketConnection {
21    socket: fasync::Socket,
22    send_buffer: VecDeque<Vec<u8>>,
23}
24
25impl SocketConnection {
26    const MAX_QUEUED_PACKETS: usize = 32;
27
28    pub fn new(socket: zx::Socket) -> Self {
29        Self {
30            socket: fasync::Socket::from_socket(socket),
31            send_buffer: VecDeque::with_capacity(Self::MAX_QUEUED_PACKETS),
32        }
33    }
34
35    pub fn into_zx_socket(self) -> zx::Socket {
36        self.socket.into_zx_socket()
37    }
38}
39
40impl Connection for SocketConnection {
41    fn closed(&self) -> Pin<Box<dyn Future<Output = Result<(), zx::Status>> + Send + 'static>> {
42        let handle = match self.socket.as_ref().duplicate_handle(zx::Rights::SAME_RIGHTS) {
43            Ok(h) => h,
44            Err(e) => return Box::pin(futures::future::ready(Err(e))),
45        };
46        Box::pin(async move {
47            let wait = fasync::OnSignals::new(handle, zx::Signals::SOCKET_PEER_CLOSED);
48            wait.map_ok(|_| ()).await
49        })
50    }
51
52    fn connection_type(&self) -> ConnectionBackendType {
53        ConnectionBackendType::Socket
54    }
55
56    fn write(&self, bytes: &[u8]) -> Result<usize, zx::Status> {
57        self.socket.as_ref().write(bytes)
58    }
59
60    fn is_closed(&self) -> bool {
61        self.socket.is_closed()
62    }
63
64    fn into_fidl_channel(self: Box<Self>) -> Result<bredr::Channel, zx::Status> {
65        let socket = self.into_zx_socket();
66        Ok(bredr::Channel { socket: Some(socket), ..Default::default() })
67    }
68}
69
70impl Stream for SocketConnection {
71    type Item = Result<Vec<u8>, zx::Status>;
72
73    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
74        let mut res = Vec::<u8>::new();
75        loop {
76            break match self.socket.poll_datagram(cx, &mut res) {
77                Poll::Ready(Ok(0)) => continue,
78                Poll::Ready(Ok(_size)) => Poll::Ready(Some(Ok(res))),
79                Poll::Ready(Err(zx::Status::PEER_CLOSED)) => Poll::Ready(None),
80                Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
81                Poll::Pending => Poll::Pending,
82            };
83        }
84    }
85}
86
87impl Sink<Vec<u8>> for SocketConnection {
88    type Error = zx::Status;
89
90    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
91        let _ = Sink::poll_flush(self.as_mut(), cx)?;
92
93        if self.send_buffer.len() >= SocketConnection::MAX_QUEUED_PACKETS {
94            return Poll::Pending;
95        }
96        Poll::Ready(Ok(()))
97    }
98
99    fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
100        self.get_mut().send_buffer.push_back(item);
101        Ok(())
102    }
103
104    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
105        let this = self.get_mut();
106        use futures::io::AsyncWrite;
107        while let Some(item) = this.send_buffer.front() {
108            let res =
109                Pin::new(&mut this.socket).poll_write(cx, item).map_err(|e| e.kind().to_status());
110            match res {
111                Poll::Ready(Ok(size)) => {
112                    if size == item.len() {
113                        let _ = this.send_buffer.pop_front();
114                    } else {
115                        error!(
116                            "Partial write in SocketConnection::Sink::poll_flush: wrote {} bytes of {} byte packet.",
117                            size,
118                            item.len()
119                        );
120                        let item = this.send_buffer.front_mut().unwrap();
121                        *item = item.split_off(size);
122                    }
123                }
124                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
125                Poll::Pending => return Poll::Pending,
126            }
127        }
128        Pin::new(&mut this.socket).poll_flush(cx).map_err(|e| e.kind().to_status())
129    }
130
131    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
132        ready!(Sink::poll_flush(self.as_mut(), cx))?;
133        let this = self.get_mut();
134        use futures::io::AsyncWrite as _;
135        Pin::new(&mut this.socket).poll_close(cx).map_err(|e| e.kind().to_status())
136    }
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142    use crate::types::Channel;
143    use futures::stream::FusedStream;
144    use futures::{SinkExt, StreamExt};
145    use std::pin::pin;
146
147    #[test]
148    fn channel_sync_write() {
149        let mut exec = fasync::TestExecutor::new();
150        let (mut recv, send) = Channel::create_socket_pair();
151
152        let heart: &[u8] = &[0xF0, 0x9F, 0x92, 0x96];
153        let size = send.write(heart).expect("write to succeed");
154        assert_eq!(size, heart.len());
155
156        let mut recv_fut = recv.next();
157        match exec.run_until_stalled(&mut recv_fut) {
158            Poll::Ready(Some(Ok(bytes))) => {
159                assert_eq!(heart, &bytes);
160            }
161            x => panic!("Expected Some(Ok(bytes)) from the stream, got {x:?}"),
162        };
163    }
164
165    #[test]
166    fn channel_into_fidl() {
167        let _exec = fasync::TestExecutor::new();
168        let (remote, _local) = zx::Socket::create_datagram();
169        let conn = SocketConnection::new(remote);
170
171        let fidl_channel =
172            Box::new(conn).into_fidl_channel().expect("into_fidl_channel to succeed");
173        assert!(fidl_channel.socket.is_some());
174        assert!(fidl_channel.connection.is_none());
175    }
176
177    #[test]
178    fn channel_closed() {
179        let mut exec = fasync::TestExecutor::new();
180
181        let (recv, send) = Channel::create_socket_pair();
182
183        let closed_fut = recv.closed();
184        let mut closed_fut = pin!(closed_fut);
185
186        assert!(exec.run_until_stalled(&mut closed_fut).is_pending());
187        assert!(!recv.is_closed());
188
189        drop(send);
190
191        assert!(exec.run_until_stalled(&mut closed_fut).is_ready());
192        assert!(recv.is_closed());
193    }
194
195    #[test]
196    fn channel_sink() {
197        let mut exec = fasync::TestExecutor::new();
198        let (mut recv, mut send) = Channel::create_socket_pair();
199
200        let data = vec![0x01, 0x02, 0x03, 0x04];
201        let mut send_fut = send.send(data.clone());
202
203        // The send should complete immediately as the socket has space.
204        match exec.run_until_stalled(&mut send_fut) {
205            Poll::Ready(Ok(())) => {}
206            x => panic!("Expected Ready(Ok(())), got {:?}", x),
207        }
208
209        let mut recv_fut = recv.next();
210        match exec.run_until_stalled(&mut recv_fut) {
211            Poll::Ready(Some(Ok(bytes))) => assert_eq!(data, bytes),
212            x => panic!("Expected successful read, got {x:?}"),
213        }
214    }
215
216    #[test]
217    fn channel_stream() {
218        let mut exec = fasync::TestExecutor::new();
219        let (remote, local) = zx::Socket::create_datagram();
220        let mut recv = Channel::from_socket(remote, Channel::DEFAULT_MAX_TX).unwrap();
221        let send = local;
222
223        let mut stream_fut = recv.next();
224
225        assert!(exec.run_until_stalled(&mut stream_fut).is_pending());
226
227        let heart: &[u8] = &[0xF0, 0x9F, 0x92, 0x96];
228        let _ = send.write(heart).expect("should write successfully");
229
230        match exec.run_until_stalled(&mut stream_fut) {
231            Poll::Ready(Some(Ok(bytes))) => {
232                assert_eq!(heart.to_vec(), bytes);
233            }
234            x => panic!("Expected Some(Ok(bytes)) from the stream, got {x:?}"),
235        }
236
237        // After the sender is dropped, the stream should terminate.
238        drop(send);
239
240        let mut stream_fut = recv.next();
241        match exec.run_until_stalled(&mut stream_fut) {
242            Poll::Ready(None) => {}
243            x => panic!("Expected None from the stream after close, got {x:?}"),
244        }
245
246        // It should continue to report terminated.
247        assert!(recv.is_terminated());
248    }
249}
fuchsia_bluetooth/types/channel/socket.rs

fuchsia_bluetooth/types/channel/
socket.rs
