fuchsia_bluetooth/types/channel/
fidl_server.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::Responder;
6use fuchsia_sync::Mutex;
7use futures::channel::{mpsc, oneshot};
8use futures::future::{BoxFuture, FusedFuture};
9use futures::sink::Sink;
10use futures::stream::Stream;
11use futures::{Future, FutureExt, SinkExt, StreamExt, ready};
12use log::{trace, warn};
13
14use fidl_fuchsia_bluetooth as fidl_bt;
15use fidl_fuchsia_bluetooth_bredr as bredr;
16use fuchsia_async as fasync;
17use std::collections::VecDeque;
18use std::pin::Pin;
19use std::sync::atomic::{AtomicBool, Ordering};
20use std::sync::{Arc, OnceLock};
21use std::task::{Context, Poll};
22use zx;
23
24use super::{Connection, ConnectionBackendType};
25
26struct FlushState {
27    outstanding_packets: std::sync::atomic::AtomicUsize,
28    waker: Mutex<Option<std::task::Waker>>,
29}
30
31/// A server-side implementation of the Bluetooth channel transport using the FIDL protocol.
32pub struct FidlServerConnection {
33    /// Shared state for tracking flush status between writer and background task.
34    flush_state: Arc<FlushState>,
35    /// Stores the first terminal error encountered by background tasks.
36    terminal_error: Arc<OnceLock<zx::Status>>,
37    /// Channel to send data to the background task (outgoing data).
38    send_tx: Mutex<mpsc::Sender<Vec<u8>>>,
39    /// Channel to receive data from the background task (incoming data).
40    recv_rx: mpsc::Receiver<Vec<u8>>,
41    /// The background task that handles FIDL requests.
42    _task: fasync::Task<()>,
43    /// True if the connection is closed.
44    is_closed: Arc<AtomicBool>,
45    /// Future that completes when the connection is closed.
46    close_fut: futures::future::Shared<BoxFuture<'static, Result<(), zx::Status>>>,
47    /// Max TX packet size.
48    max_tx_size: usize,
49}
50
51impl std::fmt::Debug for FidlServerConnection {
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        f.debug_struct("FidlServerConnection")
54            .field("is_closed", &self.is_closed.load(Ordering::Relaxed))
55            .finish()
56    }
57}
58
59impl FidlServerConnection {
60    pub const SEND_BUFFER_SIZE: usize = 32;
61
62    pub fn new(request_stream: fidl_bt::ChannelRequestStream, max_tx_size: usize) -> Self {
63        // We use a channel of size SEND_BUFFER_SIZE to match client-side capacity and support back-to-back writes.
64        let (send_tx, send_rx) = mpsc::channel(Self::SEND_BUFFER_SIZE - 1);
65        let (recv_tx, recv_rx) = mpsc::channel(Self::SEND_BUFFER_SIZE);
66        let is_closed = Arc::new(AtomicBool::new(false));
67        let (close_tx, close_rx) = oneshot::channel();
68        let flush_state = Arc::new(FlushState {
69            outstanding_packets: std::sync::atomic::AtomicUsize::new(0),
70            waker: Mutex::new(None),
71        });
72
73        let task = fasync::Task::spawn(Self::background_task(
74            request_stream,
75            send_rx,
76            recv_tx,
77            is_closed.clone(),
78            close_tx,
79            flush_state.clone(),
80        ));
81
82        let close_fut = async move {
83            let _ = close_rx.await;
84            Ok(())
85        }
86        .boxed()
87        .shared();
88
89        Self {
90            flush_state,
91            terminal_error: Arc::new(OnceLock::new()),
92            send_tx: Mutex::new(send_tx),
93            recv_rx,
94            _task: task,
95            is_closed,
96            close_fut,
97            max_tx_size,
98        }
99    }
100
101    fn send_packet(&self, packet: Vec<u8>) -> Result<(), zx::Status> {
102        trace!("FidlServerConnection: Enqueuing outgoing packet of size {}", packet.len());
103        if packet.len() > self.max_tx_size {
104            return Err(zx::Status::OUT_OF_RANGE);
105        }
106        let mut send_tx = self.send_tx.lock();
107        let _ = self.flush_state.outstanding_packets.fetch_add(1, Ordering::Relaxed);
108        if let Err(err) = send_tx.try_send(packet) {
109            let _ = self.flush_state.outstanding_packets.fetch_sub(1, Ordering::Relaxed);
110            if err.is_full() {
111                return Err(zx::Status::SHOULD_WAIT);
112            } else {
113                return Err(zx::Status::PEER_CLOSED);
114            }
115        }
116        Ok(())
117    }
118
119    fn batch_packets(send_queue: &mut VecDeque<Vec<u8>>) -> Vec<fidl_bt::Packet> {
120        let mut batch = Vec::new();
121        let mut batch_bytes = 0;
122
123        while let Some(packet) = send_queue.front() {
124            if batch_bytes + packet.len() + super::PACKET_OVERHEAD > super::MAX_BATCH_SIZE_BYTES {
125                if batch.is_empty() {
126                    // If a single packet is larger than the limit, we still try to send it.
127                    let packet = send_queue.pop_front().unwrap();
128                    batch.push(fidl_bt::Packet { packet });
129                }
130                break;
131            }
132            let packet = send_queue.pop_front().unwrap();
133            batch_bytes += packet.len() + super::PACKET_OVERHEAD;
134            batch.push(fidl_bt::Packet { packet });
135        }
136        batch
137    }
138
139    async fn background_task(
140        mut stream: fidl_bt::ChannelRequestStream,
141        mut send_rx: mpsc::Receiver<Vec<u8>>,
142        recv_tx: mpsc::Sender<Vec<u8>>,
143        is_closed: Arc<AtomicBool>,
144        close_tx: oneshot::Sender<()>,
145        flush_state: Arc<FlushState>,
146    ) {
147        let mut pending_receive: Option<fidl_bt::ChannelReceiveResponder> = None;
148        let mut send_queue = VecDeque::<Vec<u8>>::new();
149
150        // Queue of Send requests from the FIDL client.
151        let mut incoming_data =
152            VecDeque::<(VecDeque<Vec<u8>>, fidl_bt::ChannelSend_Responder)>::new();
153        // Future for forwarding packets from Send FIDL request to recv_tx.
154        let mut incoming_data_forward_fut =
155            futures::future::Fuse::<BoxFuture<'static, Result<(), mpsc::SendError>>>::terminated();
156
157        trace!("FidlServerConnection: background task started");
158
159        loop {
160            // If we are not currently forwarding a packet, check if we have pending work.
161            if incoming_data_forward_fut.is_terminated() {
162                if let Some((packets, _)) = incoming_data.front_mut() {
163                    // Use let-else to handle the case where the current request is empty.
164                    let Some(packet) = packets.pop_front() else {
165                        // All packets for this write request are forwarded. Acknowledge it.
166                        if let Some((_, responder)) = incoming_data.pop_front() {
167                            let _ = responder.send();
168                        }
169                        continue; // Restart loop to immediately process the next request
170                    };
171
172                    // Start forwarding this packet. We clone the sender to avoid borrow checker issues.
173                    let mut tx = recv_tx.clone();
174                    incoming_data_forward_fut =
175                        (async move { tx.send(packet).await }).boxed().fuse();
176                }
177            }
178
179            // Only pull more packets from the channel if the internal queue is not full.
180            // If the queue is full, we yield pending to apply backpressure to the senders.
181            let mut outgoing_fut = if send_queue.len() < Self::SEND_BUFFER_SIZE {
182                send_rx.next().left_future()
183            } else {
184                futures::future::pending().right_future()
185            };
186
187            futures::select! {
188                forward_res = incoming_data_forward_fut => {
189                    if let Err(e) = forward_res {
190                        warn!(e:?; "FidlServerConnection: Failed to forward packet to bt-rfcomm");
191                        break;
192                    }
193                }
194                request = stream.next() => {
195                    let Some(item) = request else {
196                        warn!("FidlServerConnection: FIDL request stream closed");
197                        break;
198                    };
199                    let Ok(request) = item else {
200                        warn!("FIDL request stream error: {:?}", item.unwrap_err());
201                        break;
202                    };
203                    match request {
204                        fidl_bt::ChannelRequest::Send_ { packets, responder } => {
205                            trace!("FidlServerConnection: Received Send_ request with {} packets", packets.len());
206                            let packet_data: Vec<Vec<u8>> = packets.into_iter().map(|p| p.packet).collect();
207                            incoming_data.push_back((VecDeque::from(packet_data), responder));
208                        }
209                        fidl_bt::ChannelRequest::Receive { responder } => {
210                            trace!("FidlServerConnection: Received Receive request, send_queue len: {}", send_queue.len());
211                            if !send_queue.is_empty() {
212                                let packets = Self::batch_packets(&mut send_queue);
213                                let _ = responder.send(&packets);
214                            } else {
215                                if let Some(_old) = pending_receive.replace(responder) {
216                                    warn!("Multiple outstanding Receive requests are not allowed!");
217                                    break;
218                                }
219                            }
220                        }
221                        fidl_bt::ChannelRequest::WatchChannelParameters { responder } => {
222                            warn!("FidlServerConnection: got WatchChannelParameters request, which is currently not handled");
223                            responder.drop_without_shutdown();
224                        }
225                        other => {
226                            warn!("Unknown FIDL method received: {other:?}");
227                        }
228                    }
229                }
230                outgoing = outgoing_fut => {
231                    let Some(data) = outgoing else {
232                        warn!("FidlServerConnection: send_rx closed");
233                        break;
234                    };
235                    send_queue.push_back(data);
236                    let mut processed = 1;
237                    while send_queue.len() < Self::SEND_BUFFER_SIZE {
238                        match send_rx.try_next() {
239                            Ok(Some(packet)) => {
240                                send_queue.push_back(packet);
241                                processed += 1;
242                            }
243                            _ => break,
244                        }
245                    }
246                    let outstanding = flush_state.outstanding_packets.fetch_sub(processed, Ordering::Relaxed) - processed;
247                    if outstanding == 0 {
248                        if let Some(waker) = flush_state.waker.lock().take() {
249                            waker.wake();
250                        }
251                    }
252                    if let Some(resp) = pending_receive.take() {
253                        let packets = Self::batch_packets(&mut send_queue);
254                        let _ = resp.send(&packets);
255                    }
256                }
257            }
258        }
259        trace!("FidlServerConnection: background task exiting");
260        is_closed.store(true, Ordering::Relaxed);
261        let _ = close_tx.send(());
262    }
263}
264
265impl Connection for FidlServerConnection {
266    fn closed<'a>(&'a self) -> Pin<Box<dyn Future<Output = Result<(), zx::Status>> + 'a>> {
267        Box::pin(self.close_fut.clone())
268    }
269
270    fn connection_type(&self) -> ConnectionBackendType {
271        ConnectionBackendType::FidlServer
272    }
273
274    fn write(&self, bytes: &[u8]) -> Result<usize, zx::Status> {
275        if let Some(err) = self.terminal_error.get() {
276            return Err(*err);
277        }
278        self.send_packet(bytes.to_vec())?;
279        Ok(bytes.len())
280    }
281
282    fn is_closed(&self) -> bool {
283        self.is_closed.load(Ordering::Relaxed)
284    }
285
286    fn into_fidl_channel(self: Box<Self>) -> Result<bredr::Channel, zx::Status> {
287        Err(zx::Status::NOT_SUPPORTED)
288    }
289}
290
291impl Stream for FidlServerConnection {
292    type Item = Result<Vec<u8>, zx::Status>;
293
294    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
295        self.recv_rx.poll_next_unpin(cx).map(|opt| opt.map(Ok))
296    }
297}
298
299impl Sink<Vec<u8>> for FidlServerConnection {
300    type Error = zx::Status;
301
302    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
303        let this = self.get_mut();
304        if let Some(err) = this.terminal_error.get() {
305            return Poll::Ready(Err(*err));
306        }
307        let mut send_tx = this.send_tx.lock();
308        Pin::new(&mut *send_tx).poll_ready(cx).map_err(|_| zx::Status::PEER_CLOSED)
309    }
310
311    fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
312        if let Some(err) = self.terminal_error.get() {
313            return Err(*err);
314        }
315        self.send_packet(item)
316    }
317
318    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
319        let this = self.get_mut();
320        if let Some(err) = this.terminal_error.get() {
321            return Poll::Ready(Err(*err));
322        }
323
324        if this.flush_state.outstanding_packets.load(Ordering::Relaxed) == 0 {
325            return Poll::Ready(Ok(()));
326        }
327
328        *this.flush_state.waker.lock() = Some(cx.waker().clone());
329
330        if this.flush_state.outstanding_packets.load(Ordering::Relaxed) == 0 {
331            let _ = this.flush_state.waker.lock().take();
332            return Poll::Ready(Ok(()));
333        }
334
335        Poll::Pending
336    }
337
338    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
339        ready!(Sink::poll_flush(self.as_mut(), cx))?;
340        let this = self.get_mut();
341        this.send_tx.lock().close_channel();
342        Poll::Ready(Ok(()))
343    }
344}
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349    use crate::types::Channel;
350    use fidl::endpoints::create_proxy_and_stream;
351    use fuchsia_async as fasync;
352    use futures::stream::FusedStream;
353    use futures::{SinkExt, StreamExt};
354    use std::pin::pin;
355
356    #[test]
357    fn channel_sync_write() {
358        let mut exec = fasync::TestExecutor::new();
359        let (proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
360        let channel = Channel::from_fidl_server(stream, Channel::DEFAULT_MAX_TX);
361
362        let data = vec![1, 2, 3];
363
364        // Fill up the send buffer (32 writes).
365        for _ in 0..32 {
366            let size = channel.write(&data).expect("sync write to succeed");
367            assert_eq!(size, data.len());
368        }
369
370        // The 33rd write should fail with SHOULD_WAIT because the buffer is full.
371        let result = channel.write(&data);
372        assert_eq!(result, Err(zx::Status::SHOULD_WAIT));
373
374        // Check the stream (client end) for the request.
375        let stream_fut = proxy.receive();
376        let mut stream_fut = pin!(stream_fut);
377
378        // Run executor to let background task process the send.
379        let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
380
381        match exec.run_until_stalled(&mut stream_fut) {
382            Poll::Ready(Ok(packets)) => {
383                assert_eq!(packets.len(), 32);
384                for packet in packets {
385                    assert_eq!(packet.packet, data);
386                }
387            }
388            x => panic!("Expected packets from Receive, got {:?}", x),
389        }
390
391        // And now we should be able to write again because one packet was processed and acknowledged.
392        let size = channel.write(&data).expect("sync write to succeed again");
393        assert_eq!(size, data.len());
394    }
395
396    #[test]
397    fn channel_write_too_large() {
398        let mut exec = fasync::TestExecutor::new();
399        let (_proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
400        let mut channel = Channel::from_fidl_server(stream, 10);
401
402        // Test Connection::write
403        let data = vec![1; 11];
404        let result = channel.write(&data);
405        assert_eq!(result, Err(zx::Status::OUT_OF_RANGE));
406
407        // Test SinkExt::send
408        let mut send_fut = channel.send(data);
409        let result = exec.run_until_stalled(&mut send_fut);
410        assert_eq!(result, Poll::Ready(Err(zx::Status::OUT_OF_RANGE)));
411    }
412
413    #[test]
414    fn channel_into_fidl() {
415        let _exec = fasync::TestExecutor::new();
416        let (_proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
417        let conn = FidlServerConnection::new(stream, Channel::DEFAULT_MAX_TX);
418
419        let result = Box::new(conn).into_fidl_channel();
420        assert_eq!(result.unwrap_err(), zx::Status::NOT_SUPPORTED);
421    }
422
423    #[test]
424    fn channel_closed() {
425        let mut exec = fasync::TestExecutor::new_with_fake_time();
426        let (proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
427        let channel = Channel::from_fidl_server(stream, Channel::DEFAULT_MAX_TX);
428
429        let mut closed_fut = channel.closed();
430        assert!(exec.run_until_stalled(&mut closed_fut).is_pending());
431        assert!(!channel.is_closed());
432
433        drop(proxy); // Drop the client proxy, closing the channel.
434
435        // Let the background task run to detect the closure.
436        let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
437
438        assert!(exec.run_until_stalled(&mut closed_fut).is_ready());
439        assert!(channel.is_closed());
440    }
441
442    #[test]
443    fn channel_sink() {
444        let mut exec = fasync::TestExecutor::new();
445        let (proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
446        let mut channel = Channel::from_fidl_server(stream, Channel::DEFAULT_MAX_TX);
447
448        let data = vec![1, 2, 3];
449
450        // Simulate client calling Receive.
451        let receive_fut = proxy.receive();
452        let mut receive_fut = pin!(receive_fut);
453
454        // Initially it should be pending because no data has been sent.
455        assert!(exec.run_until_stalled(&mut receive_fut).is_pending());
456
457        // Send data from server side.
458        let mut send_fut = channel.send(data.clone());
459
460        // It should be ready immediately because TestExecutor drives tasks to completion.
461        assert!(exec.run_until_stalled(&mut send_fut).is_ready());
462
463        // And the receive should also be ready with the data!
464        match exec.run_until_stalled(&mut receive_fut) {
465            Poll::Ready(Ok(packets)) => {
466                assert_eq!(packets.len(), 1);
467                assert_eq!(packets[0].packet, data);
468            }
469            x => panic!("Expected packets from Receive, got {:?}", x),
470        }
471
472        // Force proxy to stay alive until the end.
473        let _ = proxy;
474    }
475
476    #[test]
477    fn channel_stream() {
478        let mut exec = fasync::TestExecutor::new();
479        let (proxy, stream) = create_proxy_and_stream::<fidl_bt::ChannelMarker>();
480        let mut channel = Channel::from_fidl_server(stream, Channel::DEFAULT_MAX_TX);
481
482        let data = vec![4, 5, 6];
483
484        // Simulate client calling Send_
485        let send_fut = proxy.send_(&[fidl_bt::Packet { packet: data.clone() }]);
486        let mut send_fut = pin!(send_fut);
487
488        // Let the background task run
489        let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
490
491        assert!(exec.run_until_stalled(&mut send_fut).is_ready());
492
493        let mut next_fut = channel.next();
494        match exec.run_until_stalled(&mut next_fut) {
495            Poll::Ready(Some(Ok(received))) => {
496                assert_eq!(received, data);
497            }
498            x => panic!("Expected data from stream, got {:?}", x),
499        }
500
501        // After the client is dropped, the stream should terminate.
502        drop(proxy);
503
504        let _ = exec.run_until_stalled(&mut futures::future::pending::<()>());
505
506        let mut next_fut = channel.next();
507        let Poll::Ready(None) = exec.run_until_stalled(&mut next_fut) else {
508            panic!("Expected None from the stream")
509        };
510
511        assert!(channel.is_terminated());
512    }
513}
fuchsia_bluetooth/types/channel/fidl_server.rs

fuchsia_bluetooth/types/channel/
fidl_server.rs
