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

fuchsia_bluetooth/types/channel/
socket.rs
