codify "platform" into resolution logic, server-fn/fullstack overhaul (#4596)

overhaul dioxus fullstack using a new implementation of server_fn

This PR is a complete rewrite of dioxus fullstack designed to improve compatibility with axum and to fix a number of compilation errors based on feature flag unification plumbing through macros.

packages/fullstack/src/payloads/axum_types.rs

@@ -0,0 +1,77 @@
+use super::*;
+use crate::{ClientResponse, FromResponse};
+pub use axum::extract::Json;
+use axum::response::Html;
+use dioxus_fullstack_core::{RequestError, ServerFnError};
+use futures::StreamExt;
+use std::future::Future;
+
+impl<T: From<String>> FromResponse for Html<T> {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        async move {
+            let content = res.text().await?;
+            Ok(Html(content.into()))
+        }
+    }
+}
+
+impl<T> IntoRequest for Json<T>
+where
+    T: Serialize + 'static,
+{
+    fn into_request(self, request: ClientRequest) -> impl Future<Output = ClientResult> + 'static {
+        async move { request.send_json(&self.0).await }
+    }
+}
+
+impl<T: DeserializeOwned> FromResponse for Json<T> {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        async move {
+            let data = res.json::<T>().await?;
+            Ok(Json(data))
+        }
+    }
+}
+
+/// Implementation of `FromResponse` for `axum::response::Response`.
+///
+/// This allows converting a `ClientResponse` (from a client-side HTTP request)
+/// into an `axum::Response` for server-side handling. The response's status,
+/// headers, and body are transferred from the client response to the axum response.
+impl FromResponse for axum::response::Response {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        async move {
+            let parts = res.make_parts();
+            let body = axum::body::Body::from_stream(res.bytes_stream());
+            let response = axum::response::Response::from_parts(parts, body);
+            Ok(response)
+        }
+    }
+}
+
+/// Implementation of `IntoRequest` for `axum::extract::Request`.
+///
+/// This allows converting an `axum::Request` (from server-side extraction)
+/// into a `ClientRequest` that can be sent as an HTTP request. The request's
+/// headers and body are transferred from the axum request to the client request.
+impl IntoRequest for axum::extract::Request {
+    fn into_request(
+        self,
+        mut request: ClientRequest,
+    ) -> impl Future<Output = Result<ClientResponse, RequestError>> + 'static {
+        async move {
+            let (parts, body) = self.into_parts();
+
+            for (key, value) in &parts.headers {
+                request = request.header(key, value)?;
+            }
+
+            request
+                .send_body_stream(
+                    body.into_data_stream()
+                        .map(|res| res.map_err(|_| StreamingError::Failed)),
+                )
+                .await
+        }
+    }
+}

packages/fullstack/src/payloads/cbor.rs

@@ -0,0 +1,160 @@
+use axum::{
+    body::Bytes,
+    extract::{rejection::BytesRejection, FromRequest, Request},
+    http::{header, HeaderMap, HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+};
+use serde::{de::DeserializeOwned, Serialize};
+
+/// CBOR Extractor / Response.
+///
+/// When used as an extractor, it can deserialize request bodies into some type that
+/// implements [`serde::Deserialize`]. The request will be rejected (and a [`CborRejection`] will
+/// be returned) if:
+///
+/// - The request doesn't have a `Content-Type: application/cbor` (or similar) header.
+/// - The body doesn't contain syntactically valid CBOR.
+/// - The body contains syntactically valid CBOR but it couldn't be deserialized into the target type.
+/// - Buffering the request body fails.
+///
+/// ⚠️ Since parsing CBOR requires consuming the request body, the `Cbor` extractor must be
+/// *last* if there are multiple extractors in a handler.
+/// See ["the order of extractors"][order-of-extractors]
+///
+/// [order-of-extractors]: crate::extract#the-order-of-extractors
+#[must_use]
+pub struct Cbor<T>(pub T);
+
+/// Check if the request has a valid CBOR content type header.
+///
+/// This function validates that the `Content-Type` header is set to `application/cbor`
+/// or a compatible CBOR media type (including subtypes with `+cbor` suffix).
+fn is_valid_cbor_content_type(headers: &HeaderMap) -> bool {
+    let Some(content_type) = headers.get(header::CONTENT_TYPE) else {
+        return false;
+    };
+
+    let Ok(content_type) = content_type.to_str() else {
+        return false;
+    };
+
+    let Ok(mime) = content_type.parse::<mime::Mime>() else {
+        return false;
+    };
+
+    let is_cbor_content_type = mime.type_() == "application"
+        && (mime.subtype() == "cbor" || mime.suffix().is_some_and(|name| name == "cbor"));
+
+    is_cbor_content_type
+}
+
+impl<S, T> FromRequest<S> for Cbor<T>
+where
+    S: Send + Sync,
+    T: DeserializeOwned,
+{
+    type Rejection = CborRejection;
+
+    /// Extract a CBOR payload from the request body.
+    ///
+    /// This implementation validates the content type and deserializes the CBOR data.
+    /// Returns a `CborRejection` if validation or deserialization fails.
+    async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
+        if !is_valid_cbor_content_type(req.headers()) {
+            return Err(CborRejection::MissingCborContentType);
+        }
+        let bytes = Bytes::from_request(req, state).await?;
+        let value =
+            ciborium::from_reader(&bytes as &[u8]).map_err(|_| CborRejection::FailedToParseCbor)?;
+        Ok(Cbor(value))
+    }
+}
+
+impl<T> IntoResponse for Cbor<T>
+where
+    T: Serialize,
+{
+    /// Convert the CBOR payload into an HTTP response.
+    ///
+    /// This serializes the inner value to CBOR format and sets the appropriate
+    /// `Content-Type: application/cbor` header. Returns a 500 Internal Server Error
+    /// if serialization fails.
+    fn into_response(self) -> Response {
+        let mut buf = Vec::new();
+        match ciborium::into_writer(&self.0, &mut buf) {
+            Err(_) => (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                [(
+                    header::CONTENT_TYPE,
+                    HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref()),
+                )],
+                "Failed to serialize to CBOR".to_string(),
+            )
+                .into_response(),
+            Ok(()) => (
+                [(
+                    header::CONTENT_TYPE,
+                    HeaderValue::from_static("application/cbor"),
+                )],
+                buf,
+            )
+                .into_response(),
+        }
+    }
+}
+
+impl<T> From<T> for Cbor<T> {
+    /// Create a `Cbor<T>` from the inner value.
+    ///
+    /// This is a convenience constructor that wraps any value in the `Cbor` struct.
+    fn from(inner: T) -> Self {
+        Self(inner)
+    }
+}
+
+impl<T> Cbor<T>
+where
+    T: DeserializeOwned,
+{
+    /// Construct a `Cbor<T>` from a byte slice.
+    ///
+    /// This method attempts to deserialize the provided bytes as CBOR data.
+    /// Returns a `CborRejection` if deserialization fails.
+    pub fn from_bytes(bytes: &[u8]) -> Result<Self, CborRejection> {
+        ciborium::de::from_reader(bytes)
+            .map(Cbor)
+            .map_err(|_| CborRejection::FailedToParseCbor)
+    }
+}
+
+/// Rejection type for CBOR extraction failures.
+///
+/// This enum represents the various ways that CBOR extraction can fail.
+/// It implements `IntoResponse` to provide appropriate HTTP responses for each error type.
+#[derive(thiserror::Error, Debug)]
+pub enum CborRejection {
+    /// The request is missing the required `Content-Type: application/cbor` header.
+    #[error("Expected request with `Content-Type: application/cbor`")]
+    MissingCborContentType,
+
+    /// Failed to parse the request body as valid CBOR.
+    #[error("Invalid CBOR data")]
+    FailedToParseCbor,
+
+    /// Failed to read the request body bytes.
+    #[error(transparent)]
+    BytesRejection(#[from] BytesRejection),
+}
+
+impl IntoResponse for CborRejection {
+    fn into_response(self) -> Response {
+        use CborRejection::*;
+        match self {
+            MissingCborContentType => {
+                (StatusCode::UNSUPPORTED_MEDIA_TYPE, self.to_string()).into_response()
+            }
+            FailedToParseCbor => (StatusCode::BAD_REQUEST, self.to_string()).into_response(),
+            BytesRejection(rejection) => rejection.into_response(),
+        }
+    }
+}

packages/fullstack/src/payloads/files.rs

@@ -0,0 +1,371 @@
+use super::*;
+use axum_core::extract::Request;
+use dioxus_fullstack_core::RequestError;
+use dioxus_html::FileData;
+
+#[cfg(feature = "server")]
+use std::path::Path;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// A payload for uploading files using streams.
+///
+/// The `FileUpload` struct allows you to upload files by streaming their data. It can be constructed
+/// from a stream of bytes and can be sent as part of an HTTP request. This is particularly useful for
+/// handling large files without loading them entirely into memory.
+///
+/// On the web, this uses the `ReadableStream` API to stream file data.
+pub struct FileStream {
+    data: Option<FileData>,
+    name: String,
+    size: Option<u64>,
+    content_type: Option<String>,
+    #[cfg(feature = "server")]
+    server_body: Option<axum_core::body::BodyDataStream>,
+
+    // For downloaded files...
+    #[allow(clippy::type_complexity)]
+    client_body: Option<Pin<Box<dyn Stream<Item = Result<Bytes, StreamingError>> + Send>>>,
+}
+
+impl FileStream {
+    /// Get the name of the file.
+    pub fn file_name(&self) -> &str {
+        &self.name
+    }
+
+    /// Get the size of the file, if known.
+    pub fn size(&self) -> Option<u64> {
+        self.size
+    }
+
+    /// Get the content type of the file, if available.
+    pub fn content_type(&self) -> Option<&str> {
+        self.content_type.as_deref()
+    }
+
+    /// Return the underlying body stream, assuming the `FileStream` was created by a server request.
+    #[cfg(feature = "server")]
+    pub fn body_mut(&mut self) -> Option<&mut axum_core::body::BodyDataStream> {
+        self.server_body.as_mut()
+    }
+
+    /// Create a new `FileStream` from a file path. This is only available on the server.
+    #[cfg(feature = "server")]
+    pub async fn from_path(file: impl AsRef<Path>) -> Result<Self, std::io::Error> {
+        Self::from_path_buf(file.as_ref()).await
+    }
+
+    #[cfg(feature = "server")]
+    async fn from_path_buf(file: &Path) -> Result<Self, std::io::Error> {
+        let metadata = file.metadata()?;
+        let contents = tokio::fs::File::open(&file).await?;
+        let mime = dioxus_asset_resolver::native::get_mime_from_ext(
+            file.extension().and_then(|s| s.to_str()),
+        );
+        let size = metadata.len();
+        let name = file
+            .file_name()
+            .and_then(|s| s.to_str())
+            .unwrap_or("file")
+            .to_string();
+
+        // Convert the tokio file into an async byte stream
+        let reader_stream = tokio_util::io::ReaderStream::new(contents);
+
+        // Attempt to construct a BodyDataStream from the reader stream.
+        // Many axum_core versions provide a `from_stream` or similar constructor.
+        let body = axum_core::body::Body::from_stream(reader_stream).into_data_stream();
+
+        Ok(Self {
+            data: None,
+            name,
+            size: Some(size),
+            content_type: Some(mime.to_string()),
+            #[cfg(feature = "server")]
+            server_body: Some(body),
+            client_body: None,
+        })
+    }
+
+    /// Create a new `FileStream` from raw components.
+    ///
+    /// This is meant to be used on the server where a file might not even exist but you still want
+    /// to stream it to the client as a download.
+    #[cfg(feature = "server")]
+    pub fn from_raw(
+        name: String,
+        size: Option<u64>,
+        content_type: String,
+        body: axum_core::body::BodyDataStream,
+    ) -> Self {
+        Self {
+            data: None,
+            name,
+            size,
+            content_type: Some(content_type),
+            #[cfg(feature = "server")]
+            server_body: Some(body),
+            client_body: None,
+        }
+    }
+}
+
+impl IntoRequest for FileStream {
+    #[allow(unreachable_code)]
+    fn into_request(
+        self,
+        #[allow(unused_mut)] mut builder: ClientRequest,
+    ) -> impl Future<Output = ClientResult> + 'static {
+        async move {
+            let Some(file_data) = self.data else {
+                return Err(RequestError::Request(
+                    "FileStream has no data to send".into(),
+                ));
+            };
+
+            #[cfg(feature = "web")]
+            if cfg!(target_arch = "wasm32") {
+                use js_sys::escape;
+                use wasm_bindgen::JsCast;
+
+                let as_file = file_data.inner().downcast_ref::<web_sys::File>().unwrap();
+                let as_blob = as_file.dyn_ref::<web_sys::Blob>().unwrap();
+                let content_type = as_blob.type_();
+                let content_length = as_blob.size().to_string();
+                let name = as_file.name();
+
+                // Set both Content-Length and X-Content-Size for compatibility with server extraction.
+                // In browsers, content-length is often overwritten, so we set X-Content-Size as well
+                // for better compatibility with dioxus-based clients.
+                return builder
+                    .header("Content-Type", content_type)?
+                    .header("Content-Length", content_length.clone())?
+                    .header("X-Content-Size", content_length)?
+                    .header(
+                        "Content-Disposition",
+                        format!("attachment; filename=\"{}\"", escape(&name)),
+                    )?
+                    .send_js_value(as_blob.clone().into())
+                    .await;
+            }
+
+            #[cfg(not(target_arch = "wasm32"))]
+            {
+                use std::ascii::escape_default;
+
+                use futures::TryStreamExt;
+
+                let content_type = self
+                    .content_type
+                    .unwrap_or_else(|| "application/octet-stream".to_string());
+                let content_length = self.size.map(|s| s.to_string());
+                let name = self.name;
+                let stream = file_data.byte_stream().map_err(|_| StreamingError::Failed);
+
+                // Ascii escape the filename to avoid issues with special characters.
+                let mut chars = vec![];
+                for byte in name.chars() {
+                    chars.extend(escape_default(byte as u8));
+                }
+                let filename = String::from_utf8(chars).map_err(|_| {
+                    RequestError::Request(
+                        "Failed to escape filename for Content-Disposition".into(),
+                    )
+                });
+
+                if let Some(length) = content_length {
+                    builder = builder.header("Content-Length", length)?;
+                }
+
+                if let Ok(filename) = filename {
+                    builder = builder.header(
+                        "Content-Disposition",
+                        format!("attachment; filename=\"{}\"", filename),
+                    )?;
+                }
+
+                return builder
+                    .header("Content-Type", content_type)?
+                    .send_body_stream(stream)
+                    .await;
+            }
+
+            unimplemented!("FileStream::into_request is only implemented for web targets");
+        }
+    }
+}
+
+impl<S> FromRequest<S> for FileStream {
+    type Rejection = ServerFnError;
+
+    fn from_request(
+        req: Request,
+        _: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        async move {
+            tracing::info!("Extracting FileUpload from request: {:?}", req);
+
+            let disposition = req.headers().get("Content-Disposition");
+            let filename = match disposition.map(|s| s.to_str()) {
+                Some(Ok(dis)) => {
+                    let content = content_disposition::parse_content_disposition(dis);
+                    let names = content.filename();
+                    match names {
+                        Some((_name, Some(filename))) => filename.to_string(),
+                        Some((name, None)) => name.to_string(),
+                        None => "file".to_string(),
+                    }
+                }
+                _ => "file".to_string(),
+            };
+
+            // Content-length is unreliable, so we use `X-Content-Size` as an indicator.
+            // For stream requests with known bodies, the browser will still set Content-Length to 0, unfortunately.
+            let size = req
+                .headers()
+                .get("X-Content-Size")
+                .and_then(|s| s.to_str().ok())
+                .and_then(|s| s.parse::<u64>().ok());
+
+            let content_type = req
+                .headers()
+                .get("Content-Type")
+                .and_then(|s| s.to_str().ok())
+                .map(|s| s.to_string());
+
+            Ok(FileStream {
+                data: None,
+                name: filename,
+                content_type,
+                size,
+                client_body: None,
+                #[cfg(feature = "server")]
+                server_body: Some(req.into_body().into_data_stream()),
+            })
+        }
+    }
+}
+
+impl FromResponse for FileStream {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        async move {
+            // Extract filename from Content-Disposition header if present.
+            let name = res
+                .headers()
+                .get("Content-Disposition")
+                .and_then(|h| h.to_str().ok())
+                .and_then(|dis| {
+                    let cd = content_disposition::parse_content_disposition(dis);
+                    cd.filename().map(|(name, _)| name.to_string())
+                })
+                .unwrap_or_else(|| "file".to_string());
+
+            // Extract content type header
+            let content_type = res
+                .headers()
+                .get("Content-Type")
+                .and_then(|h| h.to_str().ok())
+                .map(|s| s.to_string());
+
+            // Prefer the response's known content length but fall back to X-Content-Size header.
+            let size = res.content_length().or_else(|| {
+                res.headers()
+                    .get("X-Content-Size")
+                    .and_then(|h| h.to_str().ok())
+                    .and_then(|s| s.parse::<u64>().ok())
+            });
+
+            Ok(Self {
+                data: None,
+                name,
+                size,
+                content_type,
+                client_body: Some(Box::pin(res.bytes_stream())),
+                #[cfg(feature = "server")]
+                server_body: None,
+            })
+        }
+    }
+}
+
+#[cfg(feature = "server")]
+impl IntoResponse for FileStream {
+    fn into_response(self) -> axum::response::Response {
+        use axum::body::Body;
+
+        let Some(body) = self.server_body else {
+            use dioxus_fullstack_core::HttpError;
+            return HttpError::new(http::StatusCode::BAD_REQUEST, "FileStream has no body")
+                .into_response();
+        };
+
+        let mut res = axum::response::Response::new(Body::from_stream(body));
+
+        // Set relevant headers if available
+        if let Some(content_type) = &self.content_type {
+            res.headers_mut()
+                .insert("Content-Type", content_type.parse().unwrap());
+        }
+        if let Some(size) = self.size {
+            res.headers_mut()
+                .insert("Content-Length", size.to_string().parse().unwrap());
+        }
+        res.headers_mut().insert(
+            "Content-Disposition",
+            format!("attachment; filename=\"{}\"", self.name)
+                .parse()
+                .unwrap(),
+        );
+
+        res
+    }
+}
+
+impl From<FileData> for FileStream {
+    fn from(value: FileData) -> Self {
+        Self {
+            name: value.name().to_string(),
+            content_type: value.content_type().map(|s| s.to_string()),
+            size: Some(value.size()),
+            data: Some(value),
+            client_body: None,
+            #[cfg(feature = "server")]
+            server_body: None,
+        }
+    }
+}
+
+impl Stream for FileStream {
+    type Item = Result<Bytes, StreamingError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // For server-side builds, poll the server_body stream if it exists.
+        #[cfg(feature = "server")]
+        if let Some(body) = self.server_body.as_mut() {
+            return Pin::new(body)
+                .poll_next(cx)
+                .map_err(|_| StreamingError::Failed);
+        }
+
+        // For client-side builds, poll the client_body stream if it exists.
+        if let Some(body) = self.client_body.as_mut() {
+            return body.as_mut().poll_next(cx);
+        }
+
+        // Otherwise, the stream is exhausted.
+        Poll::Ready(None)
+    }
+}
+
+impl std::fmt::Debug for FileStream {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("FileStream")
+            .field("name", &self.name)
+            .field("size", &self.size)
+            .field("content_type", &self.content_type)
+            .finish()
+    }
+}

packages/fullstack/src/payloads/form.rs

@@ -0,0 +1,12 @@
+use super::*;
+
+pub use axum::extract::Form;
+
+impl<T> IntoRequest for Form<T>
+where
+    T: Serialize + 'static,
+{
+    fn into_request(self, req: ClientRequest) -> impl Future<Output = ClientResult> + 'static {
+        async move { req.send_form(&self.0).await }
+    }
+}

packages/fullstack/src/payloads/header.rs

@@ -0,0 +1,66 @@
+use super::*;
+
+pub use headers::Cookie;
+pub use headers::SetCookie;
+
+#[derive(Clone, Debug)]
+pub struct SetHeader<Data> {
+    data: Option<Data>,
+}
+
+impl<T: Header> SetHeader<T> {
+    pub fn new(
+        value: impl TryInto<HeaderValue, Error = InvalidHeaderValue>,
+    ) -> Result<Self, headers::Error> {
+        let values = value.try_into().map_err(|_| headers::Error::invalid())?;
+
+        let res = T::decode(&mut std::iter::once(&values))?;
+
+        Ok(Self { data: Some(res) })
+    }
+}
+
+impl<T: Header> IntoResponseParts for SetHeader<T> {
+    type Error = ();
+
+    fn into_response_parts(self, res: ResponseParts) -> Result<ResponseParts, Self::Error> {
+        let data = self.data.expect("SetHeader must have data to set");
+
+        let mut headers = vec![];
+        data.encode(&mut headers);
+
+        Ok(axum::response::AppendHeaders(
+            headers.into_iter().map(|value| (T::name().clone(), value)),
+        )
+        .into_response_parts(res)
+        .unwrap())
+    }
+}
+
+impl<T: Header> FromResponseParts for SetHeader<T> {
+    fn from_response_parts(parts: &mut axum::http::response::Parts) -> Result<Self, ServerFnError> {
+        let Some(header) = parts.headers.remove(T::name()) else {
+            return Ok(SetHeader { data: None });
+        };
+
+        let data = T::decode(&mut std::iter::once(&header))
+            .map_err(|_| ServerFnError::Deserialization("Failed to decode header".into()))?;
+
+        Ok(SetHeader { data: Some(data) })
+    }
+}
+
+impl<T: Header> IntoResponse for SetHeader<T> {
+    fn into_response(self) -> axum::response::Response {
+        let mut values = vec![];
+        self.data.unwrap().encode(&mut values);
+
+        let mut response = axum::response::Response::builder();
+
+        for value in values {
+            response = response.header(T::name(), value);
+        }
+
+        response.body(axum_core::body::Body::empty()).unwrap()
+    }
+}

packages/fullstack/src/payloads/msgpack.rs

@@ -0,0 +1,186 @@
+#![forbid(unsafe_code)]
+
+use axum::{
+    body::{Body, Bytes},
+    extract::{FromRequest, Request},
+    http::{header::HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+};
+use axum::{extract::rejection::BytesRejection, http, BoxError};
+use derive_more::{Deref, DerefMut, From};
+use serde::{de::DeserializeOwned, Serialize};
+
+/// MessagePack Extractor / Response.
+///
+/// When used as an extractor, it can deserialize request bodies into some type that
+/// implements [`serde::Deserialize`]. If the request body cannot be parsed, or value of the
+/// `Content-Type` header does not match any of the `application/msgpack`, `application/x-msgpack`
+/// or `application/*+msgpack` it will reject the request and return a `400 Bad Request` response.
+///
+/// When used as a response, it can serialize any type that implements [`serde::Serialize`] to
+/// `MsgPack`, and will automatically set `Content-Type: application/msgpack` header.
+#[derive(Debug, Clone, Copy, Default, Deref, DerefMut, From)]
+pub struct MsgPack<T>(pub T);
+
+impl<T, S> FromRequest<S> for MsgPack<T>
+where
+    T: DeserializeOwned,
+    S: Send + Sync,
+{
+    type Rejection = MsgPackRejection;
+
+    async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
+        if !message_pack_content_type(&req) {
+            return Err(MsgPackRejection::MissingMsgPackContentType);
+        }
+        let bytes = Bytes::from_request(req, state).await?;
+        let value = rmp_serde::from_slice(&bytes)
+            .map_err(|e| MsgPackRejection::InvalidMsgPackBody(e.into()))?;
+        Ok(MsgPack(value))
+    }
+}
+
+impl<T> IntoResponse for MsgPack<T>
+where
+    T: Serialize,
+{
+    fn into_response(self) -> Response {
+        let bytes = match rmp_serde::encode::to_vec_named(&self.0) {
+            Ok(res) => res,
+            Err(err) => {
+                return Response::builder()
+                    .status(StatusCode::INTERNAL_SERVER_ERROR)
+                    .header("Content-Type", "text/plain")
+                    .body(Body::new(err.to_string()))
+                    .unwrap();
+            }
+        };
+
+        let mut res = bytes.into_response();
+
+        res.headers_mut().insert(
+            "Content-Type",
+            HeaderValue::from_static("application/msgpack"),
+        );
+        res
+    }
+}
+
+/// MessagePack Extractor / Response.
+///
+/// When used as an extractor, it can deserialize request bodies into some type that
+/// implements [`serde::Deserialize`]. If the request body cannot be parsed, or value of the
+/// `Content-Type` header does not match any of the `application/msgpack`, `application/x-msgpack`
+/// or `application/*+msgpack` it will reject the request and return a `400 Bad Request` response.
+#[derive(Debug, Clone, Copy, Default, Deref, DerefMut, From)]
+pub struct MsgPackRaw<T>(pub T);
+
+impl<T, S> FromRequest<S> for MsgPackRaw<T>
+where
+    T: DeserializeOwned,
+    S: Send + Sync,
+{
+    type Rejection = MsgPackRejection;
+
+    async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
+        if !message_pack_content_type(&req) {
+            return Err(MsgPackRejection::MissingMsgPackContentType);
+        }
+        let bytes = Bytes::from_request(req, state).await?;
+        let value = rmp_serde::from_slice(&bytes)
+            .map_err(|e| MsgPackRejection::InvalidMsgPackBody(e.into()))?;
+        Ok(MsgPackRaw(value))
+    }
+}
+
+impl<T> IntoResponse for MsgPackRaw<T>
+where
+    T: Serialize,
+{
+    fn into_response(self) -> Response {
+        let bytes = match rmp_serde::encode::to_vec(&self.0) {
+            Ok(res) => res,
+            Err(err) => {
+                return Response::builder()
+                    .status(StatusCode::INTERNAL_SERVER_ERROR)
+                    .header("Content-Type", "text/plain")
+                    .body(Body::new(err.to_string()))
+                    .unwrap();
+            }
+        };
+
+        let mut res = bytes.into_response();
+
+        res.headers_mut().insert(
+            "Content-Type",
+            HeaderValue::from_static("application/msgpack"),
+        );
+        res
+    }
+}
+
+fn message_pack_content_type<B>(req: &Request<B>) -> bool {
+    let Some(content_type) = req.headers().get("Content-Type") else {
+        return false;
+    };
+
+    let Ok(content_type) = content_type.to_str() else {
+        return false;
+    };
+
+    match content_type {
+        "application/msgpack" => true,
+        "application/x-msgpack" => true,
+        ct if ct.starts_with("application/") && ct.ends_with("+msgpack") => true,
+        _ => false,
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum MsgPackRejection {
+    #[error("Failed to parse the request body as MsgPack: {0}")]
+    InvalidMsgPackBody(BoxError),
+
+    #[error("Expected request with `Content-Type: application/msgpack`")]
+    MissingMsgPackContentType,
+
+    #[error("Cannot have two request body extractors for a single handler")]
+    BodyAlreadyExtracted,
+
+    #[error(transparent)]
+    BytesRejection(#[from] BytesRejection),
+}
+
+impl IntoResponse for MsgPackRejection {
+    fn into_response(self) -> Response {
+        match self {
+            Self::InvalidMsgPackBody(inner) => {
+                let mut res = Response::new(Body::from(format!(
+                    "Failed to parse the request body as MsgPack: {}",
+                    inner
+                )));
+                *res.status_mut() = http::StatusCode::BAD_REQUEST;
+                res
+            }
+
+            Self::MissingMsgPackContentType => {
+                let mut res = Response::new(Body::from(
+                    "Expected request with `Content-Type: application/msgpack`",
+                ));
+                *res.status_mut() = http::StatusCode::BAD_REQUEST;
+                res
+            }
+
+            Self::BodyAlreadyExtracted => {
+                let mut res = Response::new(Body::from(
+                    "Cannot have two request body extractors for a single handler",
+                ));
+                *res.status_mut() = http::StatusCode::INTERNAL_SERVER_ERROR;
+                res
+            }
+
+            Self::BytesRejection(inner) => inner.into_response(),
+        }
+    }
+}

packages/fullstack/src/payloads/multipart.rs

@@ -0,0 +1,165 @@
+#![allow(unreachable_code)]
+
+use crate::{ClientRequest, ClientResponse, IntoRequest};
+use axum::{
+    extract::{FromRequest, Request},
+    response::{IntoResponse, Response},
+};
+use dioxus_fullstack_core::RequestError;
+use dioxus_html::{FormData, FormEvent};
+use std::{prelude::rust_2024::Future, rc::Rc};
+
+#[cfg(feature = "server")]
+use axum::extract::multipart::{Field, MultipartError};
+
+/// A streaming multipart form data handler.
+///
+/// This type makes it easy to send and receive multipart form data in a streaming fashion by directly
+/// leveraging the corresponding `dioxus_html::FormData` and `axum::extract::Multipart` types.
+///
+/// On the client, you can create a `MultipartFormData` instance by using `.into()` on a `FormData` instance.
+/// This is typically done by using the `FormEvent`'s `.data()` method.
+///
+/// On the server, you can extract a `MultipartFormData` instance by using it as an extractor in your handler function.
+/// This gives you access to axum's `Multipart` extractor, which allows you to handle the various fields
+/// and files in the multipart form data.
+///
+/// ## Axum Usage
+///
+/// Extractor that parses `multipart/form-data` requests (commonly used with file uploads).
+///
+/// ⚠️ Since extracting multipart form data from the request requires consuming the body, the
+/// `Multipart` extractor must be *last* if there are multiple extractors in a handler.
+/// See ["the order of extractors"][order-of-extractors]
+///
+/// [order-of-extractors]: crate::extract#the-order-of-extractors
+///
+/// # Large Files
+///
+/// For security reasons, by default, `Multipart` limits the request body size to 2MB.
+/// See [`DefaultBodyLimit`][default-body-limit] for how to configure this limit.
+///
+/// [default-body-limit]: crate::extract::DefaultBodyLimit
+pub struct MultipartFormData<T = ()> {
+    #[cfg(feature = "server")]
+    form: Option<axum::extract::Multipart>,
+
+    _client: Option<Rc<FormData>>,
+
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl MultipartFormData {
+    #[cfg(feature = "server")]
+    pub async fn next_field(&mut self) -> Result<Option<Field<'_>>, MultipartError> {
+        if let Some(form) = &mut self.form {
+            form.next_field().await
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+impl<S> IntoRequest for MultipartFormData<S> {
+    fn into_request(
+        self,
+        _req: ClientRequest,
+    ) -> impl Future<Output = Result<ClientResponse, RequestError>> + 'static {
+        async move {
+            // On the web, it's just easier to convert the form data into a blob and then send that
+            // blob as the body of the request. This handles setting the correct headers, wiring
+            // up file uploads as streams, and encoding the request.
+            #[cfg(feature = "web")]
+            if cfg!(target_arch = "wasm32") {
+                let data = self._client.clone().ok_or_else(|| {
+                    RequestError::Builder("Failed to get FormData from event".into())
+                })?;
+
+                fn get_form_data(data: Rc<FormData>) -> Option<wasm_bindgen::JsValue> {
+                    use wasm_bindgen::JsCast;
+                    let event: &web_sys::Event = data.downcast()?;
+                    let target = event.target()?;
+                    let form: &web_sys::HtmlFormElement = target.dyn_ref()?;
+                    let data: web_sys::FormData = web_sys::FormData::new_with_form(form).ok()?;
+                    Some(data.into())
+                }
+
+                let js_form_data = get_form_data(data).ok_or_else(|| {
+                    RequestError::Builder("Failed to get FormData from event".into())
+                })?;
+
+                return _req.send_js_value(js_form_data).await;
+            }
+
+            // On non-web platforms, we actually need to read the values out of the FormData
+            // and construct a multipart form body manually.
+            #[cfg(not(target_arch = "wasm32"))]
+            {
+                todo!("Properly implement multipart form data for non-web wasm32 clients");
+            }
+
+            unimplemented!("Non web wasm32 clients are not supported yet")
+        }
+    }
+}
+impl<S: Send + Sync + 'static, D> FromRequest<S> for MultipartFormData<D> {
+    type Rejection = Response;
+
+    #[doc = " Perform the extraction."]
+    fn from_request(
+        req: Request,
+        state: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        #[cfg(feature = "server")]
+        return async move {
+            let form = axum::extract::multipart::Multipart::from_request(req, state)
+                .await
+                .map_err(|err| err.into_response())?;
+
+            Ok(MultipartFormData {
+                form: Some(form),
+                _client: None,
+                _phantom: std::marker::PhantomData,
+            })
+        };
+
+        #[cfg(not(feature = "server"))]
+        async {
+            use dioxus_fullstack_core::HttpError;
+
+            let _ = req;
+            let _ = state;
+            Err(HttpError::new(
+                http::StatusCode::INTERNAL_SERVER_ERROR,
+                "MultipartFormData extractor is not supported on non-server builds",
+            )
+            .into_response())
+        }
+    }
+}
+
+impl<T> From<Rc<FormData>> for MultipartFormData<T> {
+    fn from(_value: Rc<FormData>) -> Self {
+        MultipartFormData {
+            #[cfg(feature = "server")]
+            form: None,
+            _client: Some(_value),
+            _phantom: std::marker::PhantomData,
+        }
+    }
+}
+
+impl<T> From<FormEvent> for MultipartFormData<T> {
+    fn from(event: FormEvent) -> Self {
+        let data = event.data();
+        MultipartFormData {
+            #[cfg(feature = "server")]
+            form: None,
+            _client: Some(data),
+            _phantom: std::marker::PhantomData,
+        }
+    }
+}
+
+unsafe impl Send for MultipartFormData {}
+unsafe impl Sync for MultipartFormData {}

packages/fullstack/src/payloads/postcard.rs

@@ -0,0 +1,117 @@
+use axum::{
+    body::Bytes,
+    extract::{rejection::BytesRejection, FromRequest},
+    http::{header, HeaderMap, StatusCode},
+    response::{IntoResponse, Response},
+};
+use postcard::{from_bytes, to_allocvec};
+use serde::{de::DeserializeOwned, Serialize};
+use std::future::Future;
+
+/// Postcard Extractor / Response.
+///
+/// When used as an extractor, it can deserialize request bodies into some type that
+/// implements [`serde::Deserialize`]. The request will be rejected (and a [`PostcardRejection`] will
+/// be returned) if:
+///
+/// - The request doesn't have a `Content-Type: application/postcard` (or similar) header.
+/// - The body doesn't contain syntactically valid Postcard.
+/// - The body contains syntactically valid Postcard but it couldn't be deserialized into the target type.
+/// - Buffering the request body fails.
+///
+/// ⚠️ Since parsing Postcard requires consuming the request body, the `Postcard` extractor must be
+/// *last* if there are multiple extractors in a handler.
+/// See ["the order of extractors"][order-of-extractors]
+///
+/// [order-of-extractors]: crate::extract#the-order-of-extractors
+pub struct Postcard<T>(pub T);
+
+#[derive(thiserror::Error, Debug)]
+pub enum PostcardRejection {
+    #[error("Expected request with `Content-Type: application/postcard`")]
+    MissingPostcardContentType,
+
+    #[error(transparent)]
+    PostcardError(#[from] postcard::Error),
+
+    #[error(transparent)]
+    Bytes(#[from] BytesRejection),
+}
+
+impl IntoResponse for PostcardRejection {
+    fn into_response(self) -> Response {
+        use PostcardRejection::*;
+        // its often easiest to implement `IntoResponse` by calling other implementations
+        match self {
+            MissingPostcardContentType => {
+                (StatusCode::UNSUPPORTED_MEDIA_TYPE, self.to_string()).into_response()
+            }
+            PostcardError(err) => (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
+            _ => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(),
+        }
+    }
+}
+
+impl<T, S> FromRequest<S> for Postcard<T>
+where
+    T: DeserializeOwned,
+    S: Send + Sync,
+{
+    type Rejection = PostcardRejection;
+
+    fn from_request(
+        req: axum::extract::Request,
+        state: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        async move {
+            if postcard_content_type(req.headers()) {
+                let bytes = Bytes::from_request(req, state).await?;
+                let value = match from_bytes(&bytes) {
+                    Ok(value) => value,
+                    Err(err) => return Err(PostcardRejection::PostcardError(err)),
+                };
+                Ok(Postcard(value))
+            } else {
+                Err(PostcardRejection::MissingPostcardContentType)
+            }
+        }
+    }
+}
+
+fn postcard_content_type(headers: &HeaderMap) -> bool {
+    let content_type = if let Some(content_type) = headers.get(header::CONTENT_TYPE) {
+        content_type
+    } else {
+        return false;
+    };
+
+    let content_type = if let Ok(content_type) = content_type.to_str() {
+        content_type
+    } else {
+        return false;
+    };
+
+    let mime = if let Ok(mime) = content_type.parse::<mime::Mime>() {
+        mime
+    } else {
+        return false;
+    };
+
+    let is_postcard_content_type = mime.type_() == "application"
+        && (mime.subtype() == "postcard" || mime.suffix().is_some_and(|name| name == "postcard"));
+
+    is_postcard_content_type
+}
+
+impl<T> IntoResponse for Postcard<T>
+where
+    T: Serialize,
+{
+    fn into_response(self) -> Response {
+        // TODO: maybe use 128 bytes cause serde is doing something like that
+        match to_allocvec(&self.0) {
+            Ok(value) => ([(header::CONTENT_TYPE, "application/postcard")], value).into_response(),
+            Err(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
+        }
+    }
+}

packages/fullstack/src/payloads/sse.rs

@@ -0,0 +1,422 @@
+use crate::{ClientResponse, FromResponse, RequestError, ServerFnError};
+#[cfg(feature = "server")]
+use axum::{
+    response::sse::{Event, KeepAlive},
+    BoxError,
+};
+use futures::io::AsyncBufReadExt;
+use futures::Stream;
+use futures::{StreamExt, TryStreamExt};
+use http::{header::CONTENT_TYPE, HeaderValue, StatusCode};
+use serde::de::DeserializeOwned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+/// A stream of Server-Sent Events (SSE) that can be used to receive events from the server.
+///
+/// This type implements `Stream` for asynchronous iteration over events.
+/// Events are automatically deserialized from JSON to the specified type `T`.
+#[allow(clippy::type_complexity)]
+pub struct ServerEvents<T> {
+    _marker: std::marker::PhantomData<fn() -> T>,
+
+    // The receiving end from the server
+    client: Option<Pin<Box<dyn Stream<Item = Result<ServerSentEvent, ServerFnError>>>>>,
+
+    #[cfg(feature = "server")]
+    keep_alive: Option<KeepAlive>,
+
+    // The actual SSE response to send to the client
+    #[cfg(feature = "server")]
+    sse: Option<axum::response::Sse<Pin<Box<dyn Stream<Item = Result<Event, BoxError>> + Send>>>>,
+}
+
+impl<T: DeserializeOwned> ServerEvents<T> {
+    /// Receives the next event from the stream, deserializing it to `T`.
+    ///
+    /// Returns `None` if the stream has ended.
+    pub async fn recv(&mut self) -> Option<Result<T, ServerFnError>> {
+        let event = self.next_event().await?;
+        match event {
+            Ok(event) => {
+                let data: Result<T, ServerFnError> =
+                    serde_json::from_str(&event.data).map_err(|err| {
+                        ServerFnError::Serialization(format!(
+                            "failed to deserialize event data: {}",
+                            err
+                        ))
+                    });
+                Some(data)
+            }
+            Err(err) => Some(Err(err)),
+        }
+    }
+}
+
+impl<T> ServerEvents<T> {
+    /// Receives the next raw event from the stream.
+    ///
+    /// Returns `None` if the stream has ended.
+    pub async fn next_event(&mut self) -> Option<Result<ServerSentEvent, ServerFnError>> {
+        self.client.as_mut()?.next().await
+    }
+}
+
+impl<T: DeserializeOwned> Stream for ServerEvents<T> {
+    type Item = Result<T, ServerFnError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let Some(client) = self.client.as_mut() else {
+            return Poll::Ready(None);
+        };
+
+        match client.as_mut().poll_next(cx) {
+            Poll::Ready(Some(Ok(event))) => {
+                let data = serde_json::from_str(&event.data).map_err(|err| {
+                    ServerFnError::Serialization(format!(
+                        "failed to deserialize event data: {}",
+                        err
+                    ))
+                });
+                Poll::Ready(Some(data))
+            }
+            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))),
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+impl<T> FromResponse for ServerEvents<T> {
+    async fn from_response(res: ClientResponse) -> Result<Self, ServerFnError> {
+        let status = res.status();
+        if status != StatusCode::OK {
+            return Err(ServerFnError::Request(RequestError::Status(
+                format!("Expected status 200 OK, got {}", status),
+                status.as_u16(),
+            )));
+        }
+
+        let content_type = res.headers().get(CONTENT_TYPE);
+        if content_type != Some(&HeaderValue::from_static(mime::TEXT_EVENT_STREAM.as_ref())) {
+            return Err(ServerFnError::Request(RequestError::Request(format!(
+                "Expected content type 'text/event-stream', got {:?}",
+                content_type
+            ))));
+        }
+
+        let mut stream = res
+            .bytes_stream()
+            .map(|result| result.map_err(std::io::Error::other))
+            .into_async_read();
+
+        let mut line_buffer = String::new();
+        let mut event_buffer = EventBuffer::new();
+
+        let stream: Pin<Box<dyn Stream<Item = Result<ServerSentEvent, ServerFnError>>>> = Box::pin(
+            async_stream::try_stream! {
+                loop {
+                    line_buffer.clear();
+                    if stream.read_line(&mut line_buffer).await.map_err(|err| ServerFnError::StreamError(err.to_string()))? == 0 {
+                        break;
+                    }
+
+                    let line = if let Some(line) = line_buffer.strip_suffix('\n') {
+                        line
+                    } else {
+                        &line_buffer
+                    };
+
+                    // dispatch
+                    if line.is_empty() {
+                        if let Some(event) = event_buffer.produce_event() {
+                            yield event;
+                        }
+                        continue;
+                    }
+
+                    // Parse line to split field name and value, applying proper trimming.
+                    let (field, value) = line.split_once(':').unwrap_or((line, ""));
+                    let value = value.strip_prefix(' ').unwrap_or(value);
+
+                    // Handle fields - these are the in SSE speci.
+                    match field {
+                        "event" => event_buffer.set_event_type(value),
+                        "data" => event_buffer.push_data(value),
+                        "id" => event_buffer.set_id(value),
+                        "retry" => {
+                            if let Ok(millis) = value.parse() {
+                                event_buffer.set_retry(Duration::from_millis(millis));
+                            }
+                        }
+                        _ => {}
+                    }
+                }
+            },
+        );
+
+        Ok(Self {
+            _marker: std::marker::PhantomData,
+            client: Some(stream),
+
+            #[cfg(feature = "server")]
+            keep_alive: None,
+
+            #[cfg(feature = "server")]
+            sse: None,
+        })
+    }
+}
+
+/// Server-Sent Event representation.
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct ServerSentEvent {
+    /// A string identifying the type of event described.
+    pub event_type: String,
+
+    /// The data field for the message.
+    pub data: String,
+
+    /// Last event ID value.
+    pub last_event_id: Option<String>,
+
+    /// Reconnection time.
+    pub retry: Option<Duration>,
+}
+
+/// Internal buffer used to accumulate lines of an SSE (Server-Sent Events) stream.
+struct EventBuffer {
+    event_type: String,
+    data: String,
+    last_event_id: Option<String>,
+    retry: Option<Duration>,
+}
+
+impl EventBuffer {
+    /// Creates fresh new [`EventBuffer`].
+    #[allow(clippy::new_without_default)]
+    fn new() -> Self {
+        Self {
+            event_type: String::new(),
+            data: String::new(),
+            last_event_id: None,
+            retry: None,
+        }
+    }
+
+    /// Produces a [`Event`], if current state allow it.
+    ///
+    /// Reset the internal state to process further data.
+    fn produce_event(&mut self) -> Option<ServerSentEvent> {
+        let event = if self.data.is_empty() {
+            None
+        } else {
+            Some(ServerSentEvent {
+                event_type: if self.event_type.is_empty() {
+                    "message".to_string()
+                } else {
+                    self.event_type.clone()
+                },
+                data: self.data.to_string(),
+                last_event_id: self.last_event_id.clone(),
+                retry: self.retry,
+            })
+        };
+
+        self.event_type.clear();
+        self.data.clear();
+
+        event
+    }
+
+    /// Set the [`Event`]'s type. Override previous value.
+    fn set_event_type(&mut self, event_type: &str) {
+        self.event_type.clear();
+        self.event_type.push_str(event_type);
+    }
+
+    /// Extends internal data with given data.
+    fn push_data(&mut self, data: &str) {
+        if !self.data.is_empty() {
+            self.data.push('\n');
+        }
+        self.data.push_str(data);
+    }
+
+    fn set_id(&mut self, id: &str) {
+        self.last_event_id = Some(id.to_string());
+    }
+
+    fn set_retry(&mut self, retry: Duration) {
+        self.retry = Some(retry);
+    }
+}
+
+#[cfg(feature = "server")]
+pub use server_impl::*;
+
+#[cfg(feature = "server")]
+mod server_impl {
+    use super::*;
+    use crate::spawn_platform;
+    use axum::response::sse::Sse;
+    use axum_core::response::IntoResponse;
+    use futures::Future;
+    use futures::SinkExt;
+    use futures::{Sink, TryStream};
+    use serde::Serialize;
+
+    impl<T: 'static> ServerEvents<T> {
+        /// Create a `ServerEvents` from a function that is given a sender to send events to the client.
+        ///
+        /// By default, we send a comment every 15 seconds to keep the connection alive.
+        pub fn new<F, R>(f: impl FnOnce(SseTx<T>) -> F + Send + 'static) -> Self
+        where
+            F: Future<Output = R> + 'static,
+            R: 'static + Send,
+        {
+            let (tx, mut rx) = futures_channel::mpsc::unbounded();
+
+            let tx = SseTx {
+                sender: tx,
+                _marker: std::marker::PhantomData,
+            };
+
+            // Spawn the user function in the background
+            spawn_platform(move || f(tx));
+
+            // Create the stream of events, mapping the incoming events to `Ok`
+            // If the user function ends, the stream will end and the connection will be closed
+            let stream = futures::stream::poll_fn(move |cx| match rx.poll_next_unpin(cx) {
+                std::task::Poll::Ready(Some(event)) => std::task::Poll::Ready(Some(
+                    Ok(event) as Result<axum::response::sse::Event, BoxError>
+                )),
+                std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
+                std::task::Poll::Pending => std::task::Poll::Pending,
+            });
+
+            let sse = Sse::new(stream.boxed());
+
+            Self {
+                _marker: std::marker::PhantomData,
+                client: None,
+                keep_alive: Some(KeepAlive::new().interval(Duration::from_secs(15))),
+                sse: Some(sse),
+            }
+        }
+
+        /// Create a `ServerEvents` from a `TryStream` of events.
+        pub fn from_stream<S>(stream: S) -> Self
+        where
+            S: TryStream<Ok = T, Error = BoxError> + Send + 'static,
+            T: Serialize,
+        {
+            let stream = stream.map_ok(|event| {
+                axum::response::sse::Event::default()
+                    .json_data(event)
+                    .expect("Failed to serialize SSE event")
+            });
+            let sse = axum::response::Sse::new(stream.boxed());
+            Self {
+                _marker: std::marker::PhantomData,
+                client: None,
+                keep_alive: Some(KeepAlive::new().interval(Duration::from_secs(15))),
+                sse: Some(sse),
+            }
+        }
+
+        /// Set the keep-alive configuration for the SSE connection.
+        ///
+        /// A `None` value will disable the default `KeepAlive` of 15 seconds.
+        pub fn with_keep_alive(mut self, keep_alive: Option<KeepAlive>) -> Self {
+            self.keep_alive = keep_alive;
+            self
+        }
+
+        /// Create a `ServerEvents` from an existing Axum `Sse` response.
+        #[allow(clippy::type_complexity)]
+        pub fn from_sse(
+            sse: Sse<Pin<Box<dyn Stream<Item = Result<Event, BoxError>> + Send>>>,
+        ) -> Self {
+            Self {
+                _marker: std::marker::PhantomData,
+                client: None,
+                keep_alive: None,
+                sse: Some(sse),
+            }
+        }
+    }
+
+    impl<T> IntoResponse for ServerEvents<T> {
+        fn into_response(self) -> axum_core::response::Response {
+            let mut sse = self
+                .sse
+                .expect("SSE should be initialized before using it as a response");
+
+            if let Some(keep_alive) = self.keep_alive {
+                sse = sse.keep_alive(keep_alive)
+            }
+
+            sse.into_response()
+        }
+    }
+
+    /// A transmitter for sending events to the SSE stream.
+    pub struct SseTx<T> {
+        sender: futures_channel::mpsc::UnboundedSender<axum::response::sse::Event>,
+        _marker: std::marker::PhantomData<fn() -> T>,
+    }
+
+    impl<T: Serialize> SseTx<T> {
+        /// Sends an event to the SSE stream.
+        pub async fn send(&mut self, event: T) -> anyhow::Result<()> {
+            let event = axum::response::sse::Event::default().json_data(event)?;
+            self.sender.unbounded_send(event)?;
+            Ok(())
+        }
+    }
+
+    impl<T> std::ops::Deref for SseTx<T> {
+        type Target = futures_channel::mpsc::UnboundedSender<axum::response::sse::Event>;
+        fn deref(&self) -> &Self::Target {
+            &self.sender
+        }
+    }
+
+    impl<T> std::ops::DerefMut for SseTx<T> {
+        fn deref_mut(&mut self) -> &mut Self::Target {
+            &mut self.sender
+        }
+    }
+
+    impl<T: Serialize> Sink<T> for SseTx<T> {
+        type Error = anyhow::Error;
+
+        fn poll_ready(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            self.sender.poll_ready_unpin(_cx).map_err(|e| e.into())
+        }
+
+        fn start_send(mut self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+            let event = axum::response::sse::Event::default().json_data(item)?;
+            self.sender.start_send(event).map_err(|e| e.into())
+        }
+
+        fn poll_flush(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            self.sender.poll_flush_unpin(_cx).map_err(|e| e.into())
+        }
+
+        fn poll_close(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            self.sender.poll_close_unpin(_cx).map_err(|e| e.into())
+        }
+    }
+}

packages/fullstack/src/payloads/stream.rs

@@ -0,0 +1,405 @@
+#![allow(clippy::type_complexity)]
+
+use crate::{
+    CborEncoding, ClientRequest, ClientResponse, Encoding, FromResponse, IntoRequest, JsonEncoding,
+    ServerFnError,
+};
+use axum::extract::{FromRequest, Request};
+use axum_core::response::IntoResponse;
+use bytes::Bytes;
+use dioxus_fullstack_core::{HttpError, RequestError};
+use futures::{Stream, StreamExt};
+use headers::{ContentType, Header};
+use send_wrapper::SendWrapper;
+use serde::{de::DeserializeOwned, Serialize};
+use std::{future::Future, marker::PhantomData, pin::Pin};
+
+pub type TextStream = Streaming<String>;
+pub type ByteStream = Streaming<Bytes>;
+pub type JsonStream<T> = Streaming<T, JsonEncoding>;
+pub type CborStream<T> = Streaming<T, CborEncoding>;
+
+/// A streaming payload.
+///
+/// ## Browser Support for Streaming Input
+///
+/// Browser fetch requests do not currently support full request duplexing, which
+/// means that that they do not begin handling responses until the full request has been sent.
+///
+/// This means that if you use a streaming input encoding, the input stream needs to
+/// end before the output will begin.
+///
+/// Streaming requests are only allowed over HTTP2 or HTTP3.
+///
+/// Also note that not all browsers support streaming bodies to servers.
+pub struct Streaming<T = String, E = ()> {
+    output_stream: Pin<Box<dyn Stream<Item = Result<T, StreamingError>> + Send>>,
+    input_stream: Pin<Box<dyn Stream<Item = Result<T, StreamingError>> + Send>>,
+    encoding: PhantomData<E>,
+}
+
+#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq, Hash)]
+pub enum StreamingError {
+    /// The streaming request was interrupted and could not be completed.
+    #[error("The streaming request was interrupted")]
+    Interrupted,
+
+    /// The stream failed to decode a chunk - possibly due to invalid data or version mismatch.
+    #[error("The stream failed to decode a chunk")]
+    Decoding,
+
+    /// The stream failed to connect or encountered an error.
+    #[error("The streaming request failed")]
+    Failed,
+}
+
+impl<T: 'static + Send, E> Streaming<T, E> {
+    /// Creates a new stream from the given stream.
+    pub fn new(value: impl Stream<Item = T> + Send + 'static) -> Self {
+        // Box and pin the incoming stream and store as a trait object
+        Self {
+            output_stream: Box::pin(futures::stream::empty()) as _,
+            input_stream: Box::pin(value.map(|item| Ok(item)))
+                as Pin<Box<dyn Stream<Item = Result<T, StreamingError>> + Send>>,
+            encoding: PhantomData,
+        }
+    }
+
+    /// Spawns a new task that produces items for the stream.
+    ///
+    /// The callback is provided an `UnboundedSender` that can be used to send items to the stream.
+    #[cfg(feature = "server")]
+    pub fn spawn<F>(
+        callback: impl FnOnce(futures_channel::mpsc::UnboundedSender<T>) -> F + Send + 'static,
+    ) -> Self
+    where
+        F: Future<Output = ()> + 'static,
+        T: Send,
+    {
+        let (tx, rx) = futures_channel::mpsc::unbounded();
+
+        crate::spawn_platform(move || callback(tx));
+
+        Self::new(rx)
+    }
+
+    /// Returns the next item in the stream, or `None` if the stream has ended.
+    pub async fn next(&mut self) -> Option<Result<T, StreamingError>> {
+        self.output_stream.as_mut().next().await
+    }
+
+    /// Consumes the wrapper, returning the inner stream.
+    pub fn into_inner(self) -> impl Stream<Item = Result<T, StreamingError>> + Send {
+        self.input_stream
+    }
+}
+
+impl<S, U> From<S> for TextStream
+where
+    S: Stream<Item = U> + Send + 'static,
+    U: Into<String>,
+{
+    fn from(value: S) -> Self {
+        Self {
+            input_stream: Box::pin(value.map(|data| Ok(data.into()))),
+            output_stream: Box::pin(futures::stream::empty()) as _,
+            encoding: PhantomData,
+        }
+    }
+}
+
+impl<S, E> From<S> for ByteStream
+where
+    S: Stream<Item = Result<Bytes, E>> + Send + 'static,
+{
+    fn from(value: S) -> Self {
+        Self {
+            input_stream: Box::pin(value.map(|data| data.map_err(|_| StreamingError::Failed))),
+            output_stream: Box::pin(futures::stream::empty()) as _,
+            encoding: PhantomData,
+        }
+    }
+}
+
+impl<T, S, U, E> From<S> for Streaming<T, E>
+where
+    S: Stream<Item = U> + Send + 'static,
+    U: Into<T>,
+    T: 'static + Send,
+    E: Encoding,
+{
+    fn from(value: S) -> Self {
+        Self {
+            input_stream: Box::pin(value.map(|data| Ok(data.into()))),
+            output_stream: Box::pin(futures::stream::empty()) as _,
+            encoding: PhantomData,
+        }
+    }
+}
+
+impl IntoResponse for Streaming<String> {
+    fn into_response(self) -> axum_core::response::Response {
+        axum::response::Response::builder()
+            .header("Content-Type", "text/plain; charset=utf-8")
+            .body(axum::body::Body::from_stream(self.input_stream))
+            .unwrap()
+    }
+}
+
+impl IntoResponse for Streaming<Bytes> {
+    fn into_response(self) -> axum_core::response::Response {
+        axum::response::Response::builder()
+            .header("Content-Type", "application/octet-stream")
+            .body(axum::body::Body::from_stream(self.input_stream))
+            .unwrap()
+    }
+}
+
+impl<T: DeserializeOwned + Serialize + 'static, E: Encoding> IntoResponse for Streaming<T, E> {
+    fn into_response(self) -> axum_core::response::Response {
+        let res = self.input_stream.map(|r| match r {
+            Ok(res) => match E::to_bytes(&res) {
+                Some(bytes) => Ok(bytes),
+                None => Err(StreamingError::Failed),
+            },
+            Err(_err) => Err(StreamingError::Failed),
+        });
+
+        axum::response::Response::builder()
+            .header("Content-Type", E::stream_content_type())
+            .body(axum::body::Body::from_stream(res))
+            .unwrap()
+    }
+}
+
+impl FromResponse for Streaming<String> {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        SendWrapper::new(async move {
+            let client_stream = Box::pin(res.bytes_stream().map(|byte| match byte {
+                Ok(bytes) => match String::from_utf8(bytes.to_vec()) {
+                    Ok(string) => Ok(string),
+                    Err(_) => Err(StreamingError::Decoding),
+                },
+                Err(_) => Err(StreamingError::Failed),
+            }));
+
+            Ok(Self {
+                output_stream: client_stream,
+                input_stream: Box::pin(futures::stream::empty()),
+                encoding: PhantomData,
+            })
+        })
+    }
+}
+
+impl FromResponse for Streaming<Bytes> {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        async move {
+            let client_stream = Box::pin(SendWrapper::new(res.bytes_stream().map(
+                |byte| match byte {
+                    Ok(bytes) => Ok(bytes),
+                    Err(_) => Err(StreamingError::Failed),
+                },
+            )));
+
+            Ok(Self {
+                output_stream: client_stream,
+                input_stream: Box::pin(futures::stream::empty()),
+                encoding: PhantomData,
+            })
+        }
+    }
+}
+
+impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding> FromResponse
+    for Streaming<T, E>
+{
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        SendWrapper::new(async move {
+            let client_stream = Box::pin(SendWrapper::new(res.bytes_stream().map(
+                |byte| match byte {
+                    Ok(bytes) => match E::from_bytes(bytes) {
+                        Some(res) => Ok(res),
+                        None => Err(StreamingError::Decoding),
+                    },
+                    Err(_) => Err(StreamingError::Failed),
+                },
+            )));
+
+            Ok(Self {
+                output_stream: client_stream,
+                input_stream: Box::pin(futures::stream::empty()),
+                encoding: PhantomData,
+            })
+        })
+    }
+}
+
+impl<S> FromRequest<S> for Streaming<String> {
+    type Rejection = ServerFnError;
+
+    fn from_request(
+        req: Request,
+        _state: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        async move {
+            let (parts, body) = req.into_parts();
+            let content_type = parts
+                .headers
+                .get("content-type")
+                .and_then(|v| v.to_str().ok())
+                .unwrap_or("");
+
+            if !content_type.starts_with("text/plain") {
+                HttpError::bad_request("Invalid content type")?;
+            }
+
+            let stream = body.into_data_stream();
+
+            Ok(Self {
+                input_stream: Box::pin(futures::stream::empty()),
+                output_stream: Box::pin(stream.map(|byte| match byte {
+                    Ok(bytes) => match String::from_utf8(bytes.to_vec()) {
+                        Ok(string) => Ok(string),
+                        Err(_) => Err(StreamingError::Decoding),
+                    },
+                    Err(_) => Err(StreamingError::Failed),
+                })),
+                encoding: PhantomData,
+            })
+        }
+    }
+}
+
+impl<S> FromRequest<S> for ByteStream {
+    type Rejection = ServerFnError;
+
+    fn from_request(
+        req: Request,
+        _state: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        async move {
+            let (parts, body) = req.into_parts();
+            let content_type = parts
+                .headers
+                .get("content-type")
+                .and_then(|v| v.to_str().ok())
+                .unwrap_or("");
+
+            if !content_type.starts_with("application/octet-stream") {
+                HttpError::bad_request("Invalid content type")?;
+            }
+
+            let stream = body.into_data_stream();
+
+            Ok(Self {
+                input_stream: Box::pin(futures::stream::empty()),
+                output_stream: Box::pin(stream.map(|byte| match byte {
+                    Ok(bytes) => Ok(bytes),
+                    Err(_) => Err(StreamingError::Failed),
+                })),
+                encoding: PhantomData,
+            })
+        }
+    }
+}
+
+impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding, S> FromRequest<S>
+    for Streaming<T, E>
+{
+    type Rejection = ServerFnError;
+
+    fn from_request(
+        req: Request,
+        _state: &S,
+    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
+        async move {
+            let (parts, body) = req.into_parts();
+            let content_type = parts
+                .headers
+                .get("content-type")
+                .and_then(|v| v.to_str().ok())
+                .unwrap_or("");
+
+            if !content_type.starts_with(E::stream_content_type()) {
+                HttpError::bad_request("Invalid content type")?;
+            }
+
+            let stream = body.into_data_stream();
+
+            Ok(Self {
+                input_stream: Box::pin(futures::stream::empty()),
+                output_stream: Box::pin(stream.map(|byte| match byte {
+                    Ok(bytes) => match E::from_bytes(bytes) {
+                        Some(res) => Ok(res),
+                        None => Err(StreamingError::Decoding),
+                    },
+                    Err(_) => Err(StreamingError::Failed),
+                })),
+                encoding: PhantomData,
+            })
+        }
+    }
+}
+
+impl IntoRequest for Streaming<String> {
+    fn into_request(
+        self,
+        builder: ClientRequest,
+    ) -> impl Future<Output = Result<ClientResponse, RequestError>> + 'static {
+        async move {
+            builder
+                .header("Content-Type", "text/plain; charset=utf-8")?
+                .send_body_stream(self.input_stream.map(|e| e.map(Bytes::from)))
+                .await
+        }
+    }
+}
+
+impl IntoRequest for ByteStream {
+    fn into_request(
+        self,
+        builder: ClientRequest,
+    ) -> impl Future<Output = Result<ClientResponse, RequestError>> + 'static {
+        async move {
+            builder
+                .header(ContentType::name(), "application/octet-stream")?
+                .send_body_stream(self.input_stream)
+                .await
+        }
+    }
+}
+
+impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding> IntoRequest
+    for Streaming<T, E>
+{
+    fn into_request(
+        self,
+        builder: ClientRequest,
+    ) -> impl Future<Output = Result<ClientResponse, RequestError>> + 'static {
+        async move {
+            builder
+                .header("Content-Type", E::stream_content_type())?
+                .send_body_stream(
+                    self.input_stream.map(|r| {
+                        r.and_then(|item| E::to_bytes(&item).ok_or(StreamingError::Failed))
+                    }),
+                )
+                .await
+        }
+    }
+}
+
+impl<T> std::fmt::Debug for Streaming<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("Streaming").finish()
+    }
+}
+
+impl<T, E: Encoding> std::fmt::Debug for Streaming<T, E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Streaming")
+            .field("encoding", &std::any::type_name::<E>())
+            .finish()
+    }
+}

packages/fullstack/src/payloads/text.rs

@@ -0,0 +1,38 @@
+use crate::{ClientResponse, FromResponse};
+use axum_core::response::{IntoResponse, Response};
+use dioxus_fullstack_core::ServerFnError;
+use send_wrapper::SendWrapper;
+use std::future::Future;
+
+/// A simple text response type.
+///
+/// The `T` parameter can be anything that converts to and from `String`, such as `Rc<str>` or `String`.
+///
+/// Unlike `Json` or plain `String`, this uses the `text/plain` content type. The `text/plain` header
+/// will be set on the request.
+pub struct Text<T>(pub T);
+
+impl<T> Text<T> {
+    /// Create a new text response.
+    pub fn new(text: T) -> Self {
+        Self(text)
+    }
+}
+
+impl<T: Into<String>> IntoResponse for Text<T> {
+    fn into_response(self) -> Response {
+        Response::builder()
+            .header("Content-Type", "text/plain; charset=utf-8")
+            .body(axum_core::body::Body::from(self.0.into()))
+            .unwrap()
+    }
+}
+
+impl<T: From<String>> FromResponse for Text<T> {
+    fn from_response(res: ClientResponse) -> impl Future<Output = Result<Self, ServerFnError>> {
+        SendWrapper::new(async move {
+            let text = res.text().await?;
+            Ok(Text::new(text.into()))
+        })
+    }
+}