server_fn/
lib.rs

1#![forbid(unsafe_code)]
2#![deny(missing_docs)]
3
4//! # Server Functions
5//!
6//! This package is based on a simple idea: sometimes it’s useful to write functions
7//! that will only run on the server, and call them from the client.
8//!
9//! If you’re creating anything beyond a toy app, you’ll need to do this all the time:
10//! reading from or writing to a database that only runs on the server, running expensive
11//! computations using libraries you don’t want to ship down to the client, accessing
12//! APIs that need to be called from the server rather than the client for CORS reasons
13//! or because you need a secret API key that’s stored on the server and definitely
14//! shouldn’t be shipped down to a user’s browser.
15//!
16//! Traditionally, this is done by separating your server and client code, and by setting
17//! up something like a REST API or GraphQL API to allow your client to fetch and mutate
18//! data on the server. This is fine, but it requires you to write and maintain your code
19//! in multiple separate places (client-side code for fetching, server-side functions to run),
20//! as well as creating a third thing to manage, which is the API contract between the two.
21//!
22//! This package provides two simple primitives that allow you instead to write co-located,
23//! isomorphic server functions. (*Co-located* means you can write them in your app code so
24//! that they are “located alongside” the client code that calls them, rather than separating
25//! the client and server sides. *Isomorphic* means you can call them from the client as if
26//! you were simply calling a function; the function call has the “same shape” on the client
27//! as it does on the server.)
28//!
29//! ### `#[server]`
30//!
31//! The [`#[server]`](../leptos/attr.server.html) macro allows you to annotate a function to
32//! indicate that it should only run on the server (i.e., when you have an `ssr` feature in your
33//! crate that is enabled).
34//!
35//! **Important**: Before calling a server function on a non-web platform, you must set the server URL by calling
36//! [`set_server_url`](crate::client::set_server_url).
37//!
38//! ```rust,ignore
39//! #[server]
40//! async fn read_posts(how_many: usize, query: String) -> Result<Vec<Posts>, ServerFnError> {
41//!   // do some server-only work here to access the database
42//!   let posts = ...;
43//!   Ok(posts)
44//! }
45//!
46//! // call the function
47//! # #[tokio::main]
48//! # async fn main() {
49//! async {
50//!   let posts = read_posts(3, "my search".to_string()).await;
51//!   log::debug!("posts = {posts:#?}");
52//! }
53//! # }
54//! ```
55//!
56//! If you call this function from the client, it will serialize the function arguments and `POST`
57//! them to the server as if they were the URL-encoded inputs in `<form method="post">`.
58//!
59//! Here’s what you need to remember:
60//! - **Server functions must be `async`.** Even if the work being done inside the function body
61//!   can run synchronously on the server, from the client’s perspective it involves an asynchronous
62//!   function call.
63//! - **Server functions must return `Result<T, ServerFnError>`.** Even if the work being done
64//!   inside the function body can’t fail, the processes of serialization/deserialization and the
65//!   network call are fallible. [`ServerFnError`] can receive generic errors.
66//! - **Server functions are part of the public API of your application.** A server function is an
67//!   ad hoc HTTP API endpoint, not a magic formula. Any server function can be accessed by any HTTP
68//!   client. You should take care to sanitize any data being returned from the function to ensure it
69//!   does not leak data that should exist only on the server.
70//! - **Server functions can’t be generic.** Because each server function creates a separate API endpoint,
71//!   it is difficult to monomorphize. As a result, server functions cannot be generic (for now?) If you need to use
72//!   a generic function, you can define a generic inner function called by multiple concrete server functions.
73//! - **Arguments and return types must be serializable.** We support a variety of different encodings,
74//!   but one way or another arguments need to be serialized to be sent to the server and deserialized
75//!   on the server, and the return type must be serialized on the server and deserialized on the client.
76//!   This means that the set of valid server function argument and return types is a subset of all
77//!   possible Rust argument and return types. (i.e., server functions are strictly more limited than
78//!   ordinary functions.)
79//!
80//! ## Server Function Encodings
81//!
82//! Server functions are designed to allow a flexible combination of input and output encodings, the set
83//! of which can be found in the [`codec`] module.
84//!
85//! Calling and handling server functions is done through the [`Protocol`] trait, which is implemented
86//! for the [`Http`] and [`Websocket`] protocols. Most server functions will use the [`Http`] protocol.
87//!
88//! When using the [`Http`] protocol, the serialization/deserialization process for server functions
89//! consists of a series of steps, each of which is represented by a different trait:
90//! 1. [`IntoReq`]: The client serializes the [`ServerFn`] argument type into an HTTP request.
91//! 2. The [`Client`] sends the request to the server.
92//! 3. [`FromReq`]: The server deserializes the HTTP request back into the [`ServerFn`] type.
93//! 4. The server calls calls [`ServerFn::run_body`] on the data.
94//! 5. [`IntoRes`]: The server serializes the [`ServerFn::Output`] type into an HTTP response.
95//! 6. The server integration applies any middleware from [`ServerFn::middlewares`] and responds to the request.
96//! 7. [`FromRes`]: The client deserializes the response back into the [`ServerFn::Output`] type.
97//!
98//! [server]: ../leptos/attr.server.html
99//! [`serde_qs`]: <https://docs.rs/serde_qs/latest/serde_qs/>
100//! [`cbor`]: <https://docs.rs/cbor/latest/cbor/>
101
102/// Implementations of the client side of the server function call.
103pub mod client;
104
105/// Implementations of the server side of the server function call.
106pub mod server;
107
108/// Encodings for arguments and results.
109pub mod codec;
110
111#[macro_use]
112/// Error types and utilities.
113pub mod error;
114/// Types to add server middleware to a server function.
115pub mod middleware;
116/// Utilities to allow client-side redirects.
117pub mod redirect;
118/// Types and traits for  for HTTP requests.
119pub mod request;
120/// Types and traits for HTTP responses.
121pub mod response;
122
123#[cfg(feature = "actix-no-default")]
124#[doc(hidden)]
125pub use ::actix_web as actix_export;
126#[cfg(feature = "axum-no-default")]
127#[doc(hidden)]
128pub use ::axum as axum_export;
129#[cfg(feature = "generic")]
130#[doc(hidden)]
131pub use ::bytes as bytes_export;
132#[cfg(feature = "generic")]
133#[doc(hidden)]
134pub use ::http as http_export;
135use base64::{engine::general_purpose::STANDARD_NO_PAD, DecodeError, Engine};
136#[cfg(feature = "bitcode")]
137pub use bitcode;
138// re-exported to make it possible to implement a custom Client without adding a separate
139// dependency on `bytes`
140pub use bytes::Bytes;
141use bytes::{BufMut, BytesMut};
142use client::Client;
143use codec::{Encoding, FromReq, FromRes, IntoReq, IntoRes};
144#[doc(hidden)]
145pub use const_format;
146#[doc(hidden)]
147pub use const_str;
148use dashmap::DashMap;
149pub use error::ServerFnError;
150#[cfg(feature = "form-redirects")]
151use error::ServerFnUrlError;
152use error::{FromServerFnError, ServerFnErrorErr};
153use futures::{pin_mut, SinkExt, Stream, StreamExt};
154use http::Method;
155use middleware::{BoxedService, Layer, Service};
156use redirect::call_redirect_hook;
157use request::Req;
158use response::{ClientRes, Res, TryRes};
159#[cfg(feature = "rkyv")]
160pub use rkyv;
161#[doc(hidden)]
162pub use serde;
163#[doc(hidden)]
164#[cfg(feature = "serde-lite")]
165pub use serde_lite;
166use server::Server;
167use std::{
168    fmt::{Debug, Display},
169    future::Future,
170    marker::PhantomData,
171    ops::{Deref, DerefMut},
172    pin::Pin,
173    sync::{Arc, LazyLock},
174};
175#[doc(hidden)]
176pub use xxhash_rust;
177
178type ServerFnServerRequest<Fn> = <<Fn as ServerFn>::Server as crate::Server<
179    <Fn as ServerFn>::Error,
180    <Fn as ServerFn>::InputStreamError,
181    <Fn as ServerFn>::OutputStreamError,
182>>::Request;
183type ServerFnServerResponse<Fn> = <<Fn as ServerFn>::Server as crate::Server<
184    <Fn as ServerFn>::Error,
185    <Fn as ServerFn>::InputStreamError,
186    <Fn as ServerFn>::OutputStreamError,
187>>::Response;
188
189/// Defines a function that runs only on the server, but can be called from the server or the client.
190///
191/// The type for which `ServerFn` is implemented is actually the type of the arguments to the function,
192/// while the function body itself is implemented in [`run_body`](ServerFn::run_body).
193///
194/// This means that `Self` here is usually a struct, in which each field is an argument to the function.
195/// In other words,
196/// ```rust,ignore
197/// #[server]
198/// pub async fn my_function(foo: String, bar: usize) -> Result<usize, ServerFnError> {
199///     Ok(foo.len() + bar)
200/// }
201/// ```
202/// should expand to
203/// ```rust,ignore
204/// #[derive(Serialize, Deserialize)]
205/// pub struct MyFunction {
206///     foo: String,
207///     bar: usize
208/// }
209///
210/// impl ServerFn for MyFunction {
211///     async fn run_body() -> Result<usize, ServerFnError> {
212///         Ok(foo.len() + bar)
213///     }
214///
215///     // etc.
216/// }
217/// ```
218pub trait ServerFn: Send + Sized {
219    /// A unique path for the server function’s API endpoint, relative to the host, including its prefix.
220    const PATH: &'static str;
221
222    /// The type of the HTTP client that will send the request from the client side.
223    ///
224    /// For example, this might be `gloo-net` in the browser, or `reqwest` for a desktop app.
225    type Client: Client<
226        Self::Error,
227        Self::InputStreamError,
228        Self::OutputStreamError,
229    >;
230
231    /// The type of the HTTP server that will send the response from the server side.
232    ///
233    /// For example, this might be `axum` or `actix-web`.
234    type Server: Server<
235        Self::Error,
236        Self::InputStreamError,
237        Self::OutputStreamError,
238    >;
239
240    /// The protocol the server function uses to communicate with the client.
241    type Protocol: Protocol<
242        Self,
243        Self::Output,
244        Self::Client,
245        Self::Server,
246        Self::Error,
247        Self::InputStreamError,
248        Self::OutputStreamError,
249    >;
250
251    /// The return type of the server function.
252    ///
253    /// This needs to be converted into `ServerResponse` on the server side, and converted
254    /// *from* `ClientResponse` when received by the client.
255    type Output: Send;
256
257    /// The type of error in the server function return.
258    /// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
259    type Error: FromServerFnError + Send + Sync;
260    /// The type of error in the server function for stream items sent from the client to the server.
261    /// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
262    type InputStreamError: FromServerFnError + Send + Sync;
263    /// The type of error in the server function for stream items sent from the server to the client.
264    /// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
265    type OutputStreamError: FromServerFnError + Send + Sync;
266
267    /// Returns [`Self::PATH`].
268    fn url() -> &'static str {
269        Self::PATH
270    }
271
272    /// Middleware that should be applied to this server function.
273    fn middlewares() -> Vec<
274        Arc<
275            dyn Layer<
276                ServerFnServerRequest<Self>,
277                ServerFnServerResponse<Self>,
278            >,
279        >,
280    > {
281        Vec::new()
282    }
283
284    /// The body of the server function. This will only run on the server.
285    fn run_body(
286        self,
287    ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send;
288
289    #[doc(hidden)]
290    fn run_on_server(
291        req: ServerFnServerRequest<Self>,
292    ) -> impl Future<Output = ServerFnServerResponse<Self>> + Send {
293        // Server functions can either be called by a real Client,
294        // or directly by an HTML <form>. If they're accessed by a <form>, default to
295        // redirecting back to the Referer.
296        #[cfg(feature = "form-redirects")]
297        let accepts_html = req
298            .accepts()
299            .map(|n| n.contains("text/html"))
300            .unwrap_or(false);
301        #[cfg(feature = "form-redirects")]
302        let mut referer = req.referer().as_deref().map(ToOwned::to_owned);
303
304        async move {
305            #[allow(unused_variables, unused_mut)]
306            // used in form redirects feature
307            let (mut res, err) =
308                Self::Protocol::run_server(req, Self::run_body)
309                    .await
310                    .map(|res| (res, None))
311                    .unwrap_or_else(|e| {
312                        let mut response =
313                            <<Self as ServerFn>::Server as crate::Server<
314                                Self::Error,
315                                Self::InputStreamError,
316                                Self::OutputStreamError,
317                            >>::Response::error_response(
318                                Self::PATH, e.ser()
319                            );
320                        let content_type =
321                    <Self::Error as FromServerFnError>::Encoder::CONTENT_TYPE;
322                        response.content_type(content_type);
323                        (response, Some(e))
324                    });
325
326            // if it accepts HTML, we'll redirect to the Referer
327            #[cfg(feature = "form-redirects")]
328            if accepts_html {
329                // if it had an error, encode that error in the URL
330                if let Some(err) = err {
331                    if let Ok(url) = ServerFnUrlError::new(Self::PATH, err)
332                        .to_url(referer.as_deref().unwrap_or("/"))
333                    {
334                        referer = Some(url.to_string());
335                    }
336                }
337                // otherwise, strip error info from referer URL, as that means it's from a previous
338                // call
339                else if let Some(referer) = referer.as_mut() {
340                    ServerFnUrlError::<Self::Error>::strip_error_info(referer)
341                }
342
343                // set the status code and Location header
344                res.redirect(referer.as_deref().unwrap_or("/"));
345            }
346
347            res
348        }
349    }
350
351    #[doc(hidden)]
352    fn run_on_client(
353        self,
354    ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send {
355        async move { Self::Protocol::run_client(Self::PATH, self).await }
356    }
357}
358
359/// The protocol that a server function uses to communicate with the client. This trait handles
360/// the server and client side of running a server function. It is implemented for the [`Http`] and
361/// [`Websocket`] protocols and can be used to implement custom protocols.
362pub trait Protocol<
363    Input,
364    Output,
365    Client,
366    Server,
367    Error,
368    InputStreamError = Error,
369    OutputStreamError = Error,
370> where
371    Server: crate::Server<Error, InputStreamError, OutputStreamError>,
372    Client: crate::Client<Error, InputStreamError, OutputStreamError>,
373{
374    /// The HTTP method used for requests.
375    const METHOD: Method;
376
377    /// Run the server function on the server. The implementation should handle deserializing the
378    /// input, running the server function, and serializing the output.
379    fn run_server<F, Fut>(
380        request: Server::Request,
381        server_fn: F,
382    ) -> impl Future<Output = Result<Server::Response, Error>> + Send
383    where
384        F: Fn(Input) -> Fut + Send,
385        Fut: Future<Output = Result<Output, Error>> + Send;
386
387    /// Run the server function on the client. The implementation should handle serializing the
388    /// input, sending the request, and deserializing the output.
389    fn run_client(
390        path: &str,
391        input: Input,
392    ) -> impl Future<Output = Result<Output, Error>> + Send;
393}
394
395/// The http protocol with specific input and output encodings for the request and response. This is
396/// the default protocol server functions use if no override is set in the server function macro
397///
398/// The http protocol accepts two generic argument that define how the input and output for a server
399/// function are turned into HTTP requests and responses. For example, [`Http<GetUrl, Json>`] will
400/// accept a Url encoded Get request and return a JSON post response.
401///
402/// # Example
403///
404/// ```rust, no_run
405/// # use server_fn_macro_default::server;
406/// use serde::{Serialize, Deserialize};
407/// use server_fn::{Http, ServerFnError, codec::{Json, GetUrl}};
408///
409/// #[derive(Debug, Clone, Serialize, Deserialize)]
410/// pub struct Message {
411///     user: String,
412///     message: String,
413/// }
414///
415/// // The http protocol can be used on any server function that accepts and returns arguments that implement
416/// // the [`IntoReq`] and [`FromRes`] traits.
417/// //
418/// // In this case, the input and output encodings are [`GetUrl`] and [`Json`], respectively which requires
419/// // the items to implement [`IntoReq<GetUrl, ...>`] and [`FromRes<Json, ...>`]. Both of those implementations
420/// // require the items to implement [`Serialize`] and [`Deserialize`].
421/// # #[cfg(feature = "browser")] {
422/// #[server(protocol = Http<GetUrl, Json>)]
423/// async fn echo_http(
424///     input: Message,
425/// ) -> Result<Message, ServerFnError> {
426///     Ok(input)
427/// }
428/// # }
429/// ```
430pub struct Http<InputProtocol, OutputProtocol>(
431    PhantomData<(InputProtocol, OutputProtocol)>,
432);
433
434impl<InputProtocol, OutputProtocol, Input, Output, Client, Server, E>
435    Protocol<Input, Output, Client, Server, E>
436    for Http<InputProtocol, OutputProtocol>
437where
438    Input: IntoReq<InputProtocol, Client::Request, E>
439        + FromReq<InputProtocol, Server::Request, E>
440        + Send,
441    Output: IntoRes<OutputProtocol, Server::Response, E>
442        + FromRes<OutputProtocol, Client::Response, E>
443        + Send,
444    E: FromServerFnError,
445    InputProtocol: Encoding,
446    OutputProtocol: Encoding,
447    Client: crate::Client<E>,
448    Server: crate::Server<E>,
449{
450    const METHOD: Method = InputProtocol::METHOD;
451
452    async fn run_server<F, Fut>(
453        request: Server::Request,
454        server_fn: F,
455    ) -> Result<Server::Response, E>
456    where
457        F: Fn(Input) -> Fut + Send,
458        Fut: Future<Output = Result<Output, E>> + Send,
459    {
460        let input = Input::from_req(request).await?;
461
462        let output = server_fn(input).await?;
463
464        let response = Output::into_res(output).await?;
465
466        Ok(response)
467    }
468
469    async fn run_client(path: &str, input: Input) -> Result<Output, E>
470    where
471        Client: crate::Client<E>,
472    {
473        // create and send request on client
474        let req = input.into_req(path, OutputProtocol::CONTENT_TYPE)?;
475        let res = Client::send(req).await?;
476
477        let status = res.status();
478        let location = res.location();
479        let has_redirect_header = res.has_redirect();
480
481        // if it returns an error status, deserialize the error using the error's decoder.
482        let res = if (400..=599).contains(&status) {
483            Err(E::de(res.try_into_bytes().await?))
484        } else {
485            // otherwise, deserialize the body as is
486            let output = Output::from_res(res).await?;
487            Ok(output)
488        }?;
489
490        // if redirected, call the redirect hook (if that's been set)
491        if (300..=399).contains(&status) || has_redirect_header {
492            call_redirect_hook(&location);
493        }
494        Ok(res)
495    }
496}
497
498/// The websocket protocol that encodes the input and output streams using a websocket connection.
499///
500/// The websocket protocol accepts two generic argument that define the input and output serialization
501/// formats. For example, [`Websocket<CborEncoding, JsonEncoding>`] would accept a stream of Cbor-encoded messages
502/// and return a stream of JSON-encoded messages.
503///
504/// # Example
505///
506/// ```rust, no_run
507/// # use server_fn_macro_default::server;
508/// # #[cfg(feature = "browser")] {
509/// use server_fn::{ServerFnError, BoxedStream, Websocket, codec::JsonEncoding};
510/// use serde::{Serialize, Deserialize};
511///
512/// #[derive(Clone, Serialize, Deserialize)]
513/// pub struct Message {
514///     user: String,
515///     message: String,
516/// }
517/// // The websocket protocol can be used on any server function that accepts and returns a [`BoxedStream`]
518/// // with items that can be encoded by the input and output encoding generics.
519/// //
520/// // In this case, the input and output encodings are [`Json`] and [`Json`], respectively which requires
521/// // the items to implement [`Serialize`] and [`Deserialize`].
522/// #[server(protocol = Websocket<JsonEncoding, JsonEncoding>)]
523/// async fn echo_websocket(
524///     input: BoxedStream<Message, ServerFnError>,
525/// ) -> Result<BoxedStream<Message, ServerFnError>, ServerFnError> {
526///     Ok(input.into())
527/// }
528/// # }
529/// ```
530pub struct Websocket<InputEncoding, OutputEncoding>(
531    PhantomData<(InputEncoding, OutputEncoding)>,
532);
533
534/// A boxed stream type that can be used with the websocket protocol.
535///
536/// You can easily convert any static type that implement [`futures::Stream`] into a [`BoxedStream`]
537/// with the [`From`] trait.
538///
539/// # Example
540///
541/// ```rust, no_run
542/// use futures::StreamExt;
543/// use server_fn::{BoxedStream, ServerFnError};
544///
545/// let stream: BoxedStream<_, ServerFnError> =
546///     futures::stream::iter(0..10).map(Result::Ok).into();
547/// ```
548pub struct BoxedStream<T, E> {
549    stream: Pin<Box<dyn Stream<Item = Result<T, E>> + Send>>,
550}
551
552impl<T, E> From<BoxedStream<T, E>>
553    for Pin<Box<dyn Stream<Item = Result<T, E>> + Send>>
554{
555    fn from(val: BoxedStream<T, E>) -> Self {
556        val.stream
557    }
558}
559
560impl<T, E> Deref for BoxedStream<T, E> {
561    type Target = Pin<Box<dyn Stream<Item = Result<T, E>> + Send>>;
562    fn deref(&self) -> &Self::Target {
563        &self.stream
564    }
565}
566
567impl<T, E> DerefMut for BoxedStream<T, E> {
568    fn deref_mut(&mut self) -> &mut Self::Target {
569        &mut self.stream
570    }
571}
572
573impl<T, E> Debug for BoxedStream<T, E> {
574    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575        f.debug_struct("BoxedStream").finish()
576    }
577}
578
579impl<T, E, S> From<S> for BoxedStream<T, E>
580where
581    S: Stream<Item = Result<T, E>> + Send + 'static,
582{
583    fn from(stream: S) -> Self {
584        BoxedStream {
585            stream: Box::pin(stream),
586        }
587    }
588}
589
590impl<
591        Input,
592        InputItem,
593        OutputItem,
594        InputEncoding,
595        OutputEncoding,
596        Client,
597        Server,
598        Error,
599        InputStreamError,
600        OutputStreamError,
601    >
602    Protocol<
603        Input,
604        BoxedStream<OutputItem, OutputStreamError>,
605        Client,
606        Server,
607        Error,
608        InputStreamError,
609        OutputStreamError,
610    > for Websocket<InputEncoding, OutputEncoding>
611where
612    Input: Deref<Target = BoxedStream<InputItem, InputStreamError>>
613        + Into<BoxedStream<InputItem, InputStreamError>>
614        + From<BoxedStream<InputItem, InputStreamError>>,
615    InputEncoding: Encodes<InputItem> + Decodes<InputItem>,
616    OutputEncoding: Encodes<OutputItem> + Decodes<OutputItem>,
617    InputStreamError: FromServerFnError + Send,
618    OutputStreamError: FromServerFnError + Send,
619    Error: FromServerFnError + Send,
620    Server: crate::Server<Error, InputStreamError, OutputStreamError>,
621    Client: crate::Client<Error, InputStreamError, OutputStreamError>,
622    OutputItem: Send + 'static,
623    InputItem: Send + 'static,
624{
625    const METHOD: Method = Method::GET;
626
627    async fn run_server<F, Fut>(
628        request: Server::Request,
629        server_fn: F,
630    ) -> Result<Server::Response, Error>
631    where
632        F: Fn(Input) -> Fut + Send,
633        Fut: Future<
634                Output = Result<
635                    BoxedStream<OutputItem, OutputStreamError>,
636                    Error,
637                >,
638            > + Send,
639    {
640        let (request_bytes, response_stream, response) =
641            request.try_into_websocket().await?;
642        let input = request_bytes.map(|request_bytes| {
643            let request_bytes = request_bytes
644                .map(|bytes| deserialize_result::<InputStreamError>(bytes))
645                .unwrap_or_else(Err);
646            match request_bytes {
647                Ok(request_bytes) => InputEncoding::decode(request_bytes)
648                    .map_err(|e| {
649                        InputStreamError::from_server_fn_error(
650                            ServerFnErrorErr::Deserialization(e.to_string()),
651                        )
652                    }),
653                Err(err) => Err(InputStreamError::de(err)),
654            }
655        });
656        let boxed = Box::pin(input)
657            as Pin<
658                Box<
659                    dyn Stream<Item = Result<InputItem, InputStreamError>>
660                        + Send,
661                >,
662            >;
663        let input = BoxedStream { stream: boxed };
664
665        let output = server_fn(input.into()).await?;
666
667        let output = output.stream.map(|output| {
668            let result = match output {
669                Ok(output) => OutputEncoding::encode(&output).map_err(|e| {
670                    OutputStreamError::from_server_fn_error(
671                        ServerFnErrorErr::Serialization(e.to_string()),
672                    )
673                    .ser()
674                }),
675                Err(err) => Err(err.ser()),
676            };
677            serialize_result(result)
678        });
679
680        Server::spawn(async move {
681            pin_mut!(response_stream);
682            pin_mut!(output);
683            while let Some(output) = output.next().await {
684                if response_stream.send(output).await.is_err() {
685                    break;
686                }
687            }
688        })?;
689
690        Ok(response)
691    }
692
693    fn run_client(
694        path: &str,
695        input: Input,
696    ) -> impl Future<
697        Output = Result<BoxedStream<OutputItem, OutputStreamError>, Error>,
698    > + Send {
699        let input = input.into();
700
701        async move {
702            let (stream, sink) = Client::open_websocket(path).await?;
703
704            // Forward the input stream to the websocket
705            Client::spawn(async move {
706                pin_mut!(input);
707                pin_mut!(sink);
708                while let Some(input) = input.stream.next().await {
709                    let result = match input {
710                        Ok(input) => {
711                            InputEncoding::encode(&input).map_err(|e| {
712                                InputStreamError::from_server_fn_error(
713                                    ServerFnErrorErr::Serialization(
714                                        e.to_string(),
715                                    ),
716                                )
717                                .ser()
718                            })
719                        }
720                        Err(err) => Err(err.ser()),
721                    };
722                    let result = serialize_result(result);
723                    if sink.send(result).await.is_err() {
724                        break;
725                    }
726                }
727            });
728
729            // Return the output stream
730            let stream = stream.map(|request_bytes| {
731                let request_bytes = request_bytes
732                    .map(|bytes| deserialize_result::<OutputStreamError>(bytes))
733                    .unwrap_or_else(Err);
734                match request_bytes {
735                    Ok(request_bytes) => OutputEncoding::decode(request_bytes)
736                        .map_err(|e| {
737                            OutputStreamError::from_server_fn_error(
738                                ServerFnErrorErr::Deserialization(
739                                    e.to_string(),
740                                ),
741                            )
742                        }),
743                    Err(err) => Err(OutputStreamError::de(err)),
744                }
745            });
746            let boxed = Box::pin(stream)
747                as Pin<
748                    Box<
749                        dyn Stream<Item = Result<OutputItem, OutputStreamError>>
750                            + Send,
751                    >,
752                >;
753            let output = BoxedStream { stream: boxed };
754            Ok(output)
755        }
756    }
757}
758
759// Serializes a Result<Bytes, Bytes> into a single Bytes instance.
760// Format: [tag: u8][content: Bytes]
761// - Tag 0: Ok variant
762// - Tag 1: Err variant
763fn serialize_result(result: Result<Bytes, Bytes>) -> Bytes {
764    match result {
765        Ok(bytes) => {
766            let mut buf = BytesMut::with_capacity(1 + bytes.len());
767            buf.put_u8(0); // Tag for Ok variant
768            buf.extend_from_slice(&bytes);
769            buf.freeze()
770        }
771        Err(bytes) => {
772            let mut buf = BytesMut::with_capacity(1 + bytes.len());
773            buf.put_u8(1); // Tag for Err variant
774            buf.extend_from_slice(&bytes);
775            buf.freeze()
776        }
777    }
778}
779
780// Deserializes a Bytes instance back into a Result<Bytes, Bytes>.
781fn deserialize_result<E: FromServerFnError>(
782    bytes: Bytes,
783) -> Result<Bytes, Bytes> {
784    if bytes.is_empty() {
785        return Err(E::from_server_fn_error(
786            ServerFnErrorErr::Deserialization("Data is empty".into()),
787        )
788        .ser());
789    }
790
791    let tag = bytes[0];
792    let content = bytes.slice(1..);
793
794    match tag {
795        0 => Ok(content),
796        1 => Err(content),
797        _ => Err(E::from_server_fn_error(ServerFnErrorErr::Deserialization(
798            "Invalid data tag".into(),
799        ))
800        .ser()), // Invalid tag
801    }
802}
803
804/// Encode format type
805pub enum Format {
806    /// Binary representation
807    Binary,
808    /// utf-8 compatible text representation
809    Text,
810}
811/// A trait for types with an associated content type.
812pub trait ContentType {
813    /// The MIME type of the data.
814    const CONTENT_TYPE: &'static str;
815}
816
817/// Data format representation
818pub trait FormatType {
819    /// The representation type
820    const FORMAT_TYPE: Format;
821
822    /// Encodes data into a string.
823    fn into_encoded_string(bytes: Bytes) -> String {
824        match Self::FORMAT_TYPE {
825            Format::Binary => STANDARD_NO_PAD.encode(bytes),
826            Format::Text => String::from_utf8(bytes.into())
827                .expect("Valid text format type with utf-8 comptabile string"),
828        }
829    }
830
831    /// Decodes string to bytes
832    fn from_encoded_string(data: &str) -> Result<Bytes, DecodeError> {
833        match Self::FORMAT_TYPE {
834            Format::Binary => {
835                STANDARD_NO_PAD.decode(data).map(|data| data.into())
836            }
837            Format::Text => Ok(Bytes::copy_from_slice(data.as_bytes())),
838        }
839    }
840}
841
842/// A trait for types that can be encoded into a bytes for a request body.
843pub trait Encodes<T>: ContentType + FormatType {
844    /// The error type that can be returned if the encoding fails.
845    type Error: Display + Debug;
846
847    /// Encodes the given value into a bytes.
848    fn encode(output: &T) -> Result<Bytes, Self::Error>;
849}
850
851/// A trait for types that can be decoded from a bytes for a response body.
852pub trait Decodes<T> {
853    /// The error type that can be returned if the decoding fails.
854    type Error: Display;
855
856    /// Decodes the given bytes into a value.
857    fn decode(bytes: Bytes) -> Result<T, Self::Error>;
858}
859
860#[cfg(feature = "ssr")]
861#[doc(hidden)]
862pub use inventory;
863
864/// Uses the `inventory` crate to initialize a map between paths and server functions.
865#[macro_export]
866macro_rules! initialize_server_fn_map {
867    ($req:ty, $res:ty) => {
868        std::sync::LazyLock::new(|| {
869            $crate::inventory::iter::<ServerFnTraitObj<$req, $res>>
870                .into_iter()
871                .map(|obj| {
872                    ((obj.path().to_string(), obj.method()), obj.clone())
873                })
874                .collect()
875        })
876    };
877}
878
879/// A list of middlewares that can be applied to a server function.
880pub type MiddlewareSet<Req, Res> = Vec<Arc<dyn Layer<Req, Res>>>;
881
882/// A trait object that allows multiple server functions that take the same
883/// request type and return the same response type to be gathered into a single
884/// collection.
885pub struct ServerFnTraitObj<Req, Res> {
886    path: &'static str,
887    method: Method,
888    handler: fn(Req) -> Pin<Box<dyn Future<Output = Res> + Send>>,
889    middleware: fn() -> MiddlewareSet<Req, Res>,
890    ser: fn(ServerFnErrorErr) -> Bytes,
891}
892
893impl<Req, Res> ServerFnTraitObj<Req, Res> {
894    /// Converts the relevant parts of a server function into a trait object.
895    pub const fn new<
896        S: ServerFn<
897            Server: crate::Server<
898                S::Error,
899                S::InputStreamError,
900                S::OutputStreamError,
901                Request = Req,
902                Response = Res,
903            >,
904        >,
905    >(
906        handler: fn(Req) -> Pin<Box<dyn Future<Output = Res> + Send>>,
907    ) -> Self
908    where
909        Req: crate::Req<
910                S::Error,
911                S::InputStreamError,
912                S::OutputStreamError,
913                WebsocketResponse = Res,
914            > + Send
915            + 'static,
916        Res: crate::TryRes<S::Error> + Send + 'static,
917    {
918        Self {
919            path: S::PATH,
920            method: S::Protocol::METHOD,
921            handler,
922            middleware: S::middlewares,
923            ser: |e| S::Error::from_server_fn_error(e).ser(),
924        }
925    }
926
927    /// The path of the server function.
928    pub fn path(&self) -> &'static str {
929        self.path
930    }
931
932    /// The HTTP method the server function expects.
933    pub fn method(&self) -> Method {
934        self.method.clone()
935    }
936
937    /// The handler for this server function.
938    pub fn handler(&self, req: Req) -> impl Future<Output = Res> + Send {
939        (self.handler)(req)
940    }
941
942    /// The set of middleware that should be applied to this function.
943    pub fn middleware(&self) -> MiddlewareSet<Req, Res> {
944        (self.middleware)()
945    }
946
947    /// Converts the server function into a boxed service.
948    pub fn boxed(self) -> BoxedService<Req, Res>
949    where
950        Self: Service<Req, Res>,
951        Req: 'static,
952        Res: 'static,
953    {
954        BoxedService::new(self.ser, self)
955    }
956}
957
958impl<Req, Res> Service<Req, Res> for ServerFnTraitObj<Req, Res>
959where
960    Req: Send + 'static,
961    Res: 'static,
962{
963    fn run(
964        &mut self,
965        req: Req,
966        _ser: fn(ServerFnErrorErr) -> Bytes,
967    ) -> Pin<Box<dyn Future<Output = Res> + Send>> {
968        let handler = self.handler;
969        Box::pin(async move { handler(req).await })
970    }
971}
972
973impl<Req, Res> Clone for ServerFnTraitObj<Req, Res> {
974    fn clone(&self) -> Self {
975        Self {
976            path: self.path,
977            method: self.method.clone(),
978            handler: self.handler,
979            middleware: self.middleware,
980            ser: self.ser,
981        }
982    }
983}
984
985#[allow(unused)] // used by server integrations
986type LazyServerFnMap<Req, Res> =
987    LazyLock<DashMap<(String, Method), ServerFnTraitObj<Req, Res>>>;
988
989#[cfg(feature = "ssr")]
990impl<Req: 'static, Res: 'static> inventory::Collect
991    for ServerFnTraitObj<Req, Res>
992{
993    #[inline]
994    fn registry() -> &'static inventory::Registry {
995        static REGISTRY: inventory::Registry = inventory::Registry::new();
996        &REGISTRY
997    }
998}
999
1000/// Axum integration.
1001#[cfg(feature = "axum-no-default")]
1002pub mod axum {
1003    use crate::{
1004        error::FromServerFnError, middleware::BoxedService, LazyServerFnMap,
1005        Protocol, Server, ServerFn, ServerFnTraitObj,
1006    };
1007    use axum::body::Body;
1008    use http::{Method, Request, Response, StatusCode};
1009    use std::future::Future;
1010
1011    static REGISTERED_SERVER_FUNCTIONS: LazyServerFnMap<
1012        Request<Body>,
1013        Response<Body>,
1014    > = initialize_server_fn_map!(Request<Body>, Response<Body>);
1015
1016    /// The axum server function backend
1017    pub struct AxumServerFnBackend;
1018
1019    impl<Error, InputStreamError, OutputStreamError>
1020        Server<Error, InputStreamError, OutputStreamError>
1021        for AxumServerFnBackend
1022    where
1023        Error: FromServerFnError + Send + Sync,
1024        InputStreamError: FromServerFnError + Send + Sync,
1025        OutputStreamError: FromServerFnError + Send + Sync,
1026    {
1027        type Request = Request<Body>;
1028        type Response = Response<Body>;
1029
1030        #[allow(unused_variables)]
1031        fn spawn(
1032            future: impl Future<Output = ()> + Send + 'static,
1033        ) -> Result<(), Error> {
1034            #[cfg(feature = "axum")]
1035            {
1036                tokio::spawn(future);
1037                Ok(())
1038            }
1039            #[cfg(not(feature = "axum"))]
1040            {
1041                Err(Error::from_server_fn_error(
1042                    crate::error::ServerFnErrorErr::Request(
1043                        "No async runtime available. You need to either \
1044                         enable the full axum feature to pull in tokio, or \
1045                         implement the `Server` trait for your async runtime \
1046                         manually."
1047                            .into(),
1048                    ),
1049                ))
1050            }
1051        }
1052    }
1053
1054    /// Explicitly register a server function. This is only necessary if you are
1055    /// running the server in a WASM environment (or a rare environment that the
1056    /// `inventory` crate won't work in.).
1057    pub fn register_explicit<T>()
1058    where
1059        T: ServerFn<
1060                Server: crate::Server<
1061                    T::Error,
1062                    T::InputStreamError,
1063                    T::OutputStreamError,
1064                    Request = Request<Body>,
1065                    Response = Response<Body>,
1066                >,
1067            > + 'static,
1068    {
1069        REGISTERED_SERVER_FUNCTIONS.insert(
1070            (T::PATH.into(), T::Protocol::METHOD),
1071            ServerFnTraitObj::new::<T>(|req| Box::pin(T::run_on_server(req))),
1072        );
1073    }
1074
1075    /// The set of all registered server function paths.
1076    pub fn server_fn_paths() -> impl Iterator<Item = (&'static str, Method)> {
1077        REGISTERED_SERVER_FUNCTIONS
1078            .iter()
1079            .map(|item| (item.path(), item.method()))
1080    }
1081
1082    /// An Axum handler that responds to a server function request.
1083    pub async fn handle_server_fn(req: Request<Body>) -> Response<Body> {
1084        let path = req.uri().path();
1085
1086        if let Some(mut service) =
1087            get_server_fn_service(path, req.method().clone())
1088        {
1089            service.run(req).await
1090        } else {
1091            Response::builder()
1092                .status(StatusCode::BAD_REQUEST)
1093                .body(Body::from(format!(
1094                    "Could not find a server function at the route {path}. \
1095                     \n\nIt's likely that either\n 1. The API prefix you \
1096                     specify in the `#[server]` macro doesn't match the \
1097                     prefix at which your server function handler is mounted, \
1098                     or \n2. You are on a platform that doesn't support \
1099                     automatic server function registration and you need to \
1100                     call ServerFn::register_explicit() on the server \
1101                     function type, somewhere in your `main` function.",
1102                )))
1103                .unwrap()
1104        }
1105    }
1106
1107    /// Returns the server function at the given path as a service that can be modified.
1108    pub fn get_server_fn_service(
1109        path: &str,
1110        method: Method,
1111    ) -> Option<BoxedService<Request<Body>, Response<Body>>> {
1112        let key = (path.into(), method);
1113        REGISTERED_SERVER_FUNCTIONS.get(&key).map(|server_fn| {
1114            let middleware = (server_fn.middleware)();
1115            let mut service = server_fn.clone().boxed();
1116            for middleware in middleware {
1117                service = middleware.layer(service);
1118            }
1119            service
1120        })
1121    }
1122}
1123
1124/// Actix integration.
1125#[cfg(feature = "actix-no-default")]
1126pub mod actix {
1127    use crate::{
1128        error::FromServerFnError, middleware::BoxedService,
1129        request::actix::ActixRequest, response::actix::ActixResponse,
1130        server::Server, LazyServerFnMap, Protocol, ServerFn, ServerFnTraitObj,
1131    };
1132    use actix_web::{web::Payload, HttpRequest, HttpResponse};
1133    use http::Method;
1134    #[doc(hidden)]
1135    pub use send_wrapper::SendWrapper;
1136    use std::future::Future;
1137
1138    static REGISTERED_SERVER_FUNCTIONS: LazyServerFnMap<
1139        ActixRequest,
1140        ActixResponse,
1141    > = initialize_server_fn_map!(ActixRequest, ActixResponse);
1142
1143    /// The actix server function backend
1144    pub struct ActixServerFnBackend;
1145
1146    impl<Error, InputStreamError, OutputStreamError>
1147        Server<Error, InputStreamError, OutputStreamError>
1148        for ActixServerFnBackend
1149    where
1150        Error: FromServerFnError + Send + Sync,
1151        InputStreamError: FromServerFnError + Send + Sync,
1152        OutputStreamError: FromServerFnError + Send + Sync,
1153    {
1154        type Request = ActixRequest;
1155        type Response = ActixResponse;
1156
1157        fn spawn(
1158            future: impl Future<Output = ()> + Send + 'static,
1159        ) -> Result<(), Error> {
1160            actix_web::rt::spawn(future);
1161            Ok(())
1162        }
1163    }
1164
1165    /// Explicitly register a server function. This is only necessary if you are
1166    /// running the server in a WASM environment (or a rare environment that the
1167    /// `inventory` crate won't work in.).
1168    pub fn register_explicit<T>()
1169    where
1170        T: ServerFn<
1171                Server: crate::Server<
1172                    T::Error,
1173                    T::InputStreamError,
1174                    T::OutputStreamError,
1175                    Request = ActixRequest,
1176                    Response = ActixResponse,
1177                >,
1178            > + 'static,
1179    {
1180        REGISTERED_SERVER_FUNCTIONS.insert(
1181            (T::PATH.into(), T::Protocol::METHOD),
1182            ServerFnTraitObj::new::<T>(|req| Box::pin(T::run_on_server(req))),
1183        );
1184    }
1185
1186    /// The set of all registered server function paths.
1187    pub fn server_fn_paths() -> impl Iterator<Item = (&'static str, Method)> {
1188        REGISTERED_SERVER_FUNCTIONS
1189            .iter()
1190            .map(|item| (item.path(), item.method()))
1191    }
1192
1193    /// An Actix handler that responds to a server function request.
1194    pub async fn handle_server_fn(
1195        req: HttpRequest,
1196        payload: Payload,
1197    ) -> HttpResponse {
1198        let path = req.uri().path();
1199        let method = req.method();
1200        if let Some(mut service) = get_server_fn_service(path, method) {
1201            service
1202                .run(ActixRequest::from((req, payload)))
1203                .await
1204                .0
1205                .take()
1206        } else {
1207            HttpResponse::BadRequest().body(format!(
1208                "Could not find a server function at the route {path}. \
1209                 \n\nIt's likely that either\n 1. The API prefix you specify \
1210                 in the `#[server]` macro doesn't match the prefix at which \
1211                 your server function handler is mounted, or \n2. You are on \
1212                 a platform that doesn't support automatic server function \
1213                 registration and you need to call \
1214                 ServerFn::register_explicit() on the server function type, \
1215                 somewhere in your `main` function.",
1216            ))
1217        }
1218    }
1219
1220    /// Returns the server function at the given path as a service that can be modified.
1221    pub fn get_server_fn_service(
1222        path: &str,
1223        method: &actix_web::http::Method,
1224    ) -> Option<BoxedService<ActixRequest, ActixResponse>> {
1225        use actix_web::http::Method as ActixMethod;
1226
1227        let method = match *method {
1228            ActixMethod::GET => Method::GET,
1229            ActixMethod::POST => Method::POST,
1230            ActixMethod::PUT => Method::PUT,
1231            ActixMethod::PATCH => Method::PATCH,
1232            ActixMethod::DELETE => Method::DELETE,
1233            ActixMethod::HEAD => Method::HEAD,
1234            ActixMethod::TRACE => Method::TRACE,
1235            ActixMethod::OPTIONS => Method::OPTIONS,
1236            ActixMethod::CONNECT => Method::CONNECT,
1237            _ => unreachable!(),
1238        };
1239        REGISTERED_SERVER_FUNCTIONS.get(&(path.into(), method)).map(
1240            |server_fn| {
1241                let middleware = (server_fn.middleware)();
1242                let mut service = server_fn.clone().boxed();
1243                for middleware in middleware {
1244                    service = middleware.layer(service);
1245                }
1246                service
1247            },
1248        )
1249    }
1250}
1251
1252/// Mocks for the server function backend types when compiling for the client.
1253pub mod mock {
1254    use std::future::Future;
1255
1256    /// A mocked server type that can be used in place of the actual server,
1257    /// when compiling for the browser.
1258    ///
1259    /// ## Panics
1260    /// This always panics if its methods are called. It is used solely to stub out the
1261    /// server type when compiling for the client.
1262    pub struct BrowserMockServer;
1263
1264    impl<Error, InputStreamError, OutputStreamError>
1265        crate::server::Server<Error, InputStreamError, OutputStreamError>
1266        for BrowserMockServer
1267    where
1268        Error: Send + 'static,
1269        InputStreamError: Send + 'static,
1270        OutputStreamError: Send + 'static,
1271    {
1272        type Request = crate::request::BrowserMockReq;
1273        type Response = crate::response::BrowserMockRes;
1274
1275        fn spawn(
1276            _: impl Future<Output = ()> + Send + 'static,
1277        ) -> Result<(), Error> {
1278            unreachable!()
1279        }
1280    }
1281}
1282
1283#[cfg(test)]
1284mod tests {
1285
1286    use super::*;
1287    use crate::codec::JsonEncoding;
1288    use serde::{Deserialize, Serialize};
1289
1290    #[derive(Debug, Serialize, Deserialize)]
1291    enum TestError {
1292        ServerFnError(ServerFnErrorErr),
1293    }
1294
1295    impl FromServerFnError for TestError {
1296        type Encoder = JsonEncoding;
1297
1298        fn from_server_fn_error(value: ServerFnErrorErr) -> Self {
1299            Self::ServerFnError(value)
1300        }
1301    }
1302    #[test]
1303    fn test_result_serialization() {
1304        // Test Ok variant
1305        let ok_result: Result<Bytes, Bytes> =
1306            Ok(Bytes::from_static(b"success data"));
1307        let serialized = serialize_result(ok_result);
1308        let deserialized = deserialize_result::<TestError>(serialized);
1309        assert!(deserialized.is_ok());
1310        assert_eq!(deserialized.unwrap(), Bytes::from_static(b"success data"));
1311
1312        // Test Err variant
1313        let err_result: Result<Bytes, Bytes> =
1314            Err(Bytes::from_static(b"error details"));
1315        let serialized = serialize_result(err_result);
1316        let deserialized = deserialize_result::<TestError>(serialized);
1317        assert!(deserialized.is_err());
1318        assert_eq!(
1319            deserialized.unwrap_err(),
1320            Bytes::from_static(b"error details")
1321        );
1322    }
1323}