fidl_next_protocol/endpoints/
server.rs

1// Copyright 2024 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
5//! FIDL protocol servers.
6
7use core::future::Future;
8use core::mem::ManuallyDrop;
9use core::num::NonZeroU32;
10use core::pin::Pin;
11use core::ptr;
12use core::task::{Context, Poll};
13
14use fidl_next_codec::{Encode, EncodeError, EncoderExt as _};
15use pin_project::pin_project;
16
17use crate::concurrency::sync::Arc;
18use crate::concurrency::sync::atomic::{AtomicI64, Ordering};
19use crate::endpoints::connection::{Connection, SendFutureOutput, SendFutureState};
20use crate::{ProtocolError, SendFuture, Transport, decode_header, encode_header};
21
22struct ServerInner<T: Transport> {
23    connection: Connection<T>,
24    epitaph: AtomicI64,
25}
26
27impl<T: Transport> ServerInner<T> {
28    const EPITAPH_NONE: i64 = i64::MAX;
29
30    fn new(shared: T::Shared) -> Self {
31        Self { connection: Connection::new(shared), epitaph: AtomicI64::new(Self::EPITAPH_NONE) }
32    }
33
34    fn close_with_epitaph(&self, epitaph: Option<i32>) {
35        if let Some(epitaph) = epitaph {
36            self.epitaph.store(epitaph as i64, Ordering::Relaxed);
37        }
38        self.connection.stop();
39    }
40
41    fn epitaph(&self) -> Option<i32> {
42        let epitaph = self.epitaph.load(Ordering::Relaxed);
43        if epitaph != Self::EPITAPH_NONE { Some(epitaph as i32) } else { None }
44    }
45}
46
47/// A server endpoint.
48pub struct Server<T: Transport> {
49    inner: Arc<ServerInner<T>>,
50}
51
52impl<T: Transport> Server<T> {
53    /// Closes the channel from the server end.
54    pub fn close(&self) {
55        self.inner.close_with_epitaph(None);
56    }
57
58    /// Closes the channel from the server end after sending an epitaph message.
59    pub fn close_with_epitaph(&self, epitaph: i32) {
60        self.inner.close_with_epitaph(Some(epitaph));
61    }
62
63    /// Send an event.
64    pub fn send_event<M>(&self, ordinal: u64, event: M) -> Result<SendFuture<'_, T>, EncodeError>
65    where
66        M: Encode<T::SendBuffer>,
67    {
68        self.inner.connection.send_message(|buffer| {
69            encode_header::<T>(buffer, 0, ordinal)?;
70            buffer.encode_next(event)
71        })
72    }
73}
74
75impl<T: Transport> Clone for Server<T> {
76    fn clone(&self) -> Self {
77        Self { inner: self.inner.clone() }
78    }
79}
80
81/// A type which handles incoming events for a server.
82///
83/// The futures returned by `on_one_way` and `on_two_way` are required to be `Send`. See
84/// `LocalServerHandler` for a version of this trait which does not require the returned futures to
85/// be `Send`.
86pub trait ServerHandler<T: Transport> {
87    /// Handles a received one-way server message.
88    ///
89    /// The server cannot handle more messages until `on_one_way` completes. If `on_one_way` may
90    /// block, perform asynchronous work, or take a long time to process a message, it should
91    /// offload work to an async task.
92    fn on_one_way(
93        &mut self,
94        ordinal: u64,
95        buffer: T::RecvBuffer,
96    ) -> impl Future<Output = Result<(), ProtocolError<T::Error>>> + Send;
97
98    /// Handles a received two-way server message.
99    ///
100    /// The server cannot handle more messages until `on_two_way` completes. If `on_two_way` may
101    /// block, perform asynchronous work, or take a long time to process a message, it should
102    /// offload work to an async task.
103    fn on_two_way(
104        &mut self,
105        ordinal: u64,
106        buffer: T::RecvBuffer,
107        responder: Responder<T>,
108    ) -> impl Future<Output = Result<(), ProtocolError<T::Error>>> + Send;
109}
110
111/// A dispatcher for a server endpoint.
112///
113/// A server dispatcher receives all of the incoming requests and dispatches them to the server
114/// handler. It acts as the message pump for the server.
115///
116/// The dispatcher must be actively polled to receive requests. If the dispatcher is not
117/// [`run`](ServerDispatcher::run), then requests will not be received.
118pub struct ServerDispatcher<T: Transport> {
119    inner: Arc<ServerInner<T>>,
120    exclusive: T::Exclusive,
121    is_terminated: bool,
122}
123
124impl<T: Transport> Drop for ServerDispatcher<T> {
125    fn drop(&mut self) {
126        if !self.is_terminated {
127            // SAFETY: We checked that the connection has not been terminated.
128            unsafe {
129                self.inner.connection.terminate(ProtocolError::Stopped);
130            }
131        }
132    }
133}
134
135impl<T: Transport> ServerDispatcher<T> {
136    /// Creates a new server from a transport.
137    pub fn new(transport: T) -> Self {
138        let (shared, exclusive) = transport.split();
139        Self { inner: Arc::new(ServerInner::new(shared)), exclusive, is_terminated: false }
140    }
141
142    /// Returns the dispatcher's server.
143    pub fn server(&self) -> Server<T> {
144        Server { inner: self.inner.clone() }
145    }
146
147    /// Runs the server with the provided handler.
148    pub async fn run<H>(mut self, mut handler: H) -> Result<H, ProtocolError<T::Error>>
149    where
150        H: ServerHandler<T>,
151    {
152        // We may assume that the connection has not been terminated because
153        // connections are only terminated by `run` and `drop`. Neither of those
154        // could have been called before this method because `run` consumes
155        // `self` and `drop` is only ever called once.
156
157        let error = loop {
158            // SAFETY: The connection has not been terminated.
159            let result = unsafe { self.run_one(&mut handler).await };
160            if let Err(error) = result {
161                break error;
162            }
163        };
164
165        // If we closed locally, we may have an epitaph to send before
166        // terminating the connection.
167        if matches!(error, ProtocolError::Stopped) {
168            if let Some(epitaph) = self.inner.epitaph() {
169                // Note that we don't care whether sending the epitaph succeeds
170                // or fails; it's best-effort.
171
172                // SAFETY: The connection has not been terminated.
173                let _ = unsafe { self.inner.connection.send_epitaph(epitaph).await };
174            }
175        }
176
177        // SAFETY: The connection has not been terminated.
178        unsafe {
179            self.inner.connection.terminate(error.clone());
180        }
181        self.is_terminated = true;
182
183        match error {
184            // We consider servers to have finished successfully if they stop
185            // themselves manually, or if the client disconnects.
186            ProtocolError::Stopped | ProtocolError::PeerClosed => Ok(handler),
187
188            // Otherwise, the server finished with an error.
189            _ => Err(error),
190        }
191    }
192
193    /// # Safety
194    ///
195    /// The connection must not be terminated.
196    async unsafe fn run_one<H>(&mut self, handler: &mut H) -> Result<(), ProtocolError<T::Error>>
197    where
198        H: ServerHandler<T>,
199    {
200        // SAFETY: The caller guaranteed that the connection is not terminated.
201        let mut buffer = unsafe { self.inner.connection.recv(&mut self.exclusive).await? };
202
203        let (txid, ordinal) =
204            decode_header::<T>(&mut buffer).map_err(ProtocolError::InvalidMessageHeader)?;
205        if let Some(txid) = NonZeroU32::new(txid) {
206            let responder = Responder { server: self.server(), txid };
207            handler.on_two_way(ordinal, buffer, responder).await?;
208        } else {
209            handler.on_one_way(ordinal, buffer).await?;
210        }
211
212        Ok(())
213    }
214}
215
216/// A responder for a two-way message.
217#[must_use = "responders close the underlying FIDL connection when dropped"]
218pub struct Responder<T: Transport> {
219    server: Server<T>,
220    txid: NonZeroU32,
221}
222
223impl<T: Transport> Drop for Responder<T> {
224    fn drop(&mut self) {
225        self.server.close();
226    }
227}
228
229impl<T: Transport> Responder<T> {
230    /// Send a response to a two-way message.
231    pub fn respond<M>(self, ordinal: u64, response: M) -> Result<RespondFuture<T>, EncodeError>
232    where
233        M: Encode<T::SendBuffer>,
234    {
235        let state = self.server.inner.connection.send_message_raw(|buffer| {
236            encode_header::<T>(buffer, self.txid.get(), ordinal)?;
237            buffer.encode_next(response)
238        })?;
239
240        let this = ManuallyDrop::new(self);
241        // SAFETY: `this` is a `ManuallyDrop` and so `server` won't be dropped
242        // twice.
243        let server = unsafe { ptr::read(&this.server) };
244
245        Ok(RespondFuture { server, state })
246    }
247}
248
249/// A future which responds to a request over a connection.
250#[must_use = "futures do nothing unless polled"]
251#[pin_project]
252pub struct RespondFuture<T: Transport> {
253    server: Server<T>,
254    #[pin]
255    state: SendFutureState<T>,
256}
257
258impl<T: Transport> Future for RespondFuture<T> {
259    type Output = SendFutureOutput<T>;
260
261    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
262        let this = self.project();
263
264        this.state.poll_send(cx, &this.server.inner.connection)
265    }
266}
fidl_next_protocol/endpoints/server.rs

fidl_next_protocol/endpoints/
server.rs
