multiple fullstack nits/cleanups before release (#4852)

* feat: `:arg` syntax for serverfn

* add better decode/encode verify message

* allow catch-all query args

* add rebind

* implement websocket frames for `Streaming`

* fix lazy init

* wip... serverfnstate

* wip...

* wip: serverfns require Send/Sync, let's fix that

* wip...

* small cleanup

* fix hang by actually spawning the tokio task

* clippy

* fix parameter catchall binding

* fix lychee

* doc tests

packages/fullstack/src/payloads/axum_types.rs

@@ -18,7 +18,7 @@ impl<T: From<String>> FromResponse for Html<T> {
 
 impl<T> IntoRequest for Json<T>
 where
-    T: Serialize + 'static,
+    T: Serialize + 'static + DeserializeOwned,
 {
     fn into_request(self, request: ClientRequest) -> impl Future<Output = ClientResult> + 'static {
         async move { request.send_json(&self.0).await }

packages/fullstack/src/payloads/form.rs

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

packages/fullstack/src/payloads/query.rs

@@ -0,0 +1,34 @@
+use std::ops::Deref;
+
+use crate::ServerFnError;
+use axum::extract::FromRequestParts;
+use http::request::Parts;
+use serde::de::DeserializeOwned;
+
+/// An extractor that deserializes query parameters into the given type `T`.
+///
+/// This uses `serde_qs` under the hood to support complex query parameter structures.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct Query<T>(pub T);
+
+impl<T, S> FromRequestParts<S> for Query<T>
+where
+    T: DeserializeOwned,
+    S: Send + Sync,
+{
+    type Rejection = ServerFnError;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        let inner: T = serde_qs::from_str(parts.uri.query().unwrap_or_default())
+            .map_err(|e| ServerFnError::Deserialization(e.to_string()))?;
+        Ok(Self(inner))
+    }
+}
+
+impl<T> Deref for Query<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}

packages/fullstack/src/payloads/stream.rs

@@ -9,18 +9,83 @@ use axum_core::response::IntoResponse;
 use bytes::Bytes;
 use dioxus_fullstack_core::{HttpError, RequestError};
 use futures::{Stream, StreamExt};
+#[cfg(feature = "server")]
+use futures_channel::mpsc::UnboundedSender;
 use headers::{ContentType, Header};
 use send_wrapper::SendWrapper;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{future::Future, marker::PhantomData, pin::Pin};
 
+/// A stream of text data.
+///
+/// # Chunking
+///
+/// Note that strings sent by the server might not arrive in the same chunking as they were sent.
+///
+/// This is because the underlying transport layer (HTTP/2 or HTTP/3) may choose to split or combine
+/// chunks for efficiency.
+///
+/// If you need to preserve individual string boundaries, consider using `ChunkedTextStream` or another
+/// encoding that preserves chunk boundaries.
 pub type TextStream = Streaming<String>;
+
+/// A stream of binary data.
+///
+/// # Chunking
+///
+/// Note that bytes sent by the server might not arrive in the same chunking as they were sent.
+/// This is because the underlying transport layer (HTTP/2 or HTTP/3) may choose to split or combine
+/// chunks for efficiency.
+///
+/// If you need to preserve individual byte boundaries, consider using `ChunkedByteStream` or another
+/// encoding that preserves chunk boundaries.
 pub type ByteStream = Streaming<Bytes>;
+
+/// A stream of JSON-encoded data.
+///
+/// # Chunking
+///
+/// Normally, it's not possible to stream JSON over HTTP because browsers are free to re-chunk
+/// data as they see fit. However, this implementation manually frames each JSON as if it were an unmasked
+/// websocket message.
+///
+/// If you need to send a stream of JSON data without framing, consider using TextStream instead and
+/// manually handling JSON buffering.
 pub type JsonStream<T> = Streaming<T, JsonEncoding>;
+
+/// A stream of Cbor-encoded data.
+///
+/// # Chunking
+///
+/// Normally, it's not possible to stream JSON over HTTP because browsers are free to re-chunk
+/// data as they see fit. However, this implementation manually frames each item as if it were an unmasked
+/// websocket message.
 pub type CborStream<T> = Streaming<T, CborEncoding>;
 
+/// A stream of manually chunked binary data.
+///
+/// This encoding preserves chunk boundaries by framing each chunk with its length, using Websocket
+/// Framing.
+pub type ChunkedByteStream = Streaming<Bytes, CborEncoding>;
+
+/// A stream of manually chunked text data.
+///
+/// This encoding preserves chunk boundaries by framing each chunk with its length, using Websocket
+/// Framing.
+pub type ChunkedTextStream = Streaming<String, CborEncoding>;
+
 /// A streaming payload.
 ///
+/// ## Frames and Chunking
+///
+/// The streaming payload sends and receives data in discrete chunks or "frames". The size is converted
+/// to hex and sent before each chunk, followed by a CRLF, the chunk data, and another CRLF.
+///
+/// This mimics actual HTTP chunked transfer encoding, but allows us to define our own framing
+/// protocol on top of it.
+///
+/// Arbitrary bytes can be encoded between these frames, but the frames do come with some overhead.
+///
 /// ## Browser Support for Streaming Input
 ///
 /// Browser fetch requests do not currently support full request duplexing, which
@@ -58,9 +123,9 @@ impl<T: 'static + Send, E> Streaming<T, E> {
     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>>,
+            output_stream: Box::pin(futures::stream::empty()) as _,
             encoding: PhantomData,
         }
     }
@@ -69,9 +134,7 @@ impl<T: 'static + Send, E> Streaming<T, E> {
     ///
     /// 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
+    pub fn spawn<F>(callback: impl FnOnce(UnboundedSender<T>) -> F + Send + 'static) -> Self
     where
         F: Future<Output = ()> + 'static,
         T: Send,
@@ -92,6 +155,17 @@ impl<T: 'static + Send, E> Streaming<T, E> {
     pub fn into_inner(self) -> impl Stream<Item = Result<T, StreamingError>> + Send {
         self.input_stream
     }
+
+    /// Creates a streaming payload from an existing stream of bytes.
+    ///
+    /// This uses the internal framing mechanism to decode the stream into items of type `T`.
+    fn from_bytes(stream: impl Stream<Item = Result<T, StreamingError>> + Send + 'static) -> Self {
+        Self {
+            input_stream: Box::pin(stream),
+            output_stream: Box::pin(futures::stream::empty()) as _,
+            encoding: PhantomData,
+        }
+    }
 }
 
 impl<S, U> From<S> for TextStream
@@ -100,11 +174,7 @@ where
     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,
-        }
+        Self::new(value.map(|data| data.into()))
     }
 }
 
@@ -129,11 +199,7 @@ where
     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,
-        }
+        Self::from_bytes(value.map(|data| Ok(data.into())))
     }
 }
 
@@ -158,7 +224,7 @@ impl IntoResponse for Streaming<Bytes> {
 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) {
+            Ok(res) => match encode_stream_frame::<T, E>(res) {
                 Some(bytes) => Ok(bytes),
                 None => Err(StreamingError::Failed),
             },
@@ -218,7 +284,7 @@ impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding> FromResponse
         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) {
+                    Ok(bytes) => match decode_stream_frame::<T, E>(bytes) {
                         Some(res) => Ok(res),
                         None => Err(StreamingError::Decoding),
                     },
@@ -330,7 +396,7 @@ impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding, S> FromReque
             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) {
+                    Ok(bytes) => match decode_stream_frame::<T, E>(bytes) {
                         Some(res) => Ok(res),
                         None => Err(StreamingError::Decoding),
                     },
@@ -380,11 +446,11 @@ impl<T: DeserializeOwned + Serialize + 'static + Send, E: Encoding> IntoRequest
         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))
-                    }),
-                )
+                .send_body_stream(self.input_stream.map(|r| {
+                    r.and_then(|item| {
+                        encode_stream_frame::<T, E>(item).ok_or(StreamingError::Failed)
+                    })
+                }))
                 .await
         }
     }
@@ -403,3 +469,113 @@ impl<T, E: Encoding> std::fmt::Debug for Streaming<T, E> {
             .finish()
     }
 }
+
+/// This function encodes a single frame of a streaming payload using the specified encoding.
+///
+/// The resulting `Bytes` object is encoded as a websocket frame, so you can send it over a streaming
+/// HTTP response or even a websocket connection.
+///
+/// Note that the packet is not masked, as it is assumed to be sent over a trusted connection.
+pub fn encode_stream_frame<T: Serialize, E: Encoding>(data: T) -> Option<Bytes> {
+    // We use full advantage of `BytesMut` here, writing a maximally full frame and then shrinking it
+    // down to size at the end.
+    //
+    // Also note we don't do any masking over this data since it's not going over an untrusted
+    // network like a websocket would.
+    //
+    // We allocate 10 extra bytes to account for framing overhead, which we'll shrink after
+    let mut bytes = vec![0u8; 10];
+
+    E::encode(data, &mut bytes)?;
+
+    let len = (bytes.len() - 10) as u64;
+    let opcode = 0x82; // FIN + binary opcode
+
+    // Write the header directly into the allocated space.
+    let offset = if len <= 125 {
+        bytes[8] = opcode;
+        bytes[9] = len as u8;
+        8
+    } else if len <= u16::MAX as u64 {
+        bytes[6] = opcode;
+        bytes[7] = 126;
+        let len_bytes = (len as u16).to_be_bytes();
+        bytes[8] = len_bytes[0];
+        bytes[9] = len_bytes[1];
+        6
+    } else {
+        bytes[0] = opcode;
+        bytes[1] = 127;
+        bytes[2..10].copy_from_slice(&len.to_be_bytes());
+        0
+    };
+
+    // Shrink down to the actual used size - is zero copy!
+    Some(Bytes::from(bytes).slice(offset..))
+}
+
+/// Decode a websocket-framed streaming payload produced by [`encode_stream_frame`].
+///
+/// This function returns `None` if the frame is invalid or cannot be decoded.
+///
+/// It cannot handle masked frames, as those are not produced by our encoding function.
+pub fn decode_stream_frame<T, E>(frame: Bytes) -> Option<T>
+where
+    E: Encoding,
+    T: DeserializeOwned,
+{
+    let data = frame.as_ref();
+
+    if data.len() < 2 {
+        return None;
+    }
+
+    let first = data[0];
+    let second = data[1];
+
+    // Require FIN with binary opcode and no RSV bits
+    let fin = first & 0x80 != 0;
+    let opcode = first & 0x0F;
+    let rsv = first & 0x70;
+    if !fin || opcode != 0x02 || rsv != 0 {
+        return None;
+    }
+
+    // Mask bit must be zero for our framing
+    if second & 0x80 != 0 {
+        return None;
+    }
+
+    let mut offset = 2usize;
+    let mut payload_len = (second & 0x7F) as usize;
+
+    if payload_len == 126 {
+        if data.len() < offset + 2 {
+            return None;
+        }
+
+        payload_len = u16::from_be_bytes([data[offset], data[offset + 1]]) as usize;
+        offset += 2;
+    } else if payload_len == 127 {
+        if data.len() < offset + 8 {
+            return None;
+        }
+
+        let mut len_bytes = [0u8; 8];
+        len_bytes.copy_from_slice(&data[offset..offset + 8]);
+        let len_u64 = u64::from_be_bytes(len_bytes);
+
+        if len_u64 > usize::MAX as u64 {
+            return None;
+        }
+
+        payload_len = len_u64 as usize;
+        offset += 8;
+    }
+
+    if data.len() < offset + payload_len {
+        return None;
+    }
+
+    E::decode(frame.slice(offset..offset + payload_len))
+}

packages/fullstack/src/payloads/websocket.rs

@@ -307,11 +307,11 @@ impl<I, O, E> Websocket<I, O, E> {
             match msg {
                 Message::Text(text) => {
                     let e: O =
-                        E::from_bytes(text.into()).ok_or_else(WebsocketError::deserialization)?;
+                        E::decode(text.into()).ok_or_else(WebsocketError::deserialization)?;
                     return Ok(e);
                 }
                 Message::Binary(bytes) => {
-                    let e: O = E::from_bytes(bytes).ok_or_else(WebsocketError::deserialization)?;
+                    let e: O = E::decode(bytes).ok_or_else(WebsocketError::deserialization)?;
                     return Ok(e);
                 }
                 Message::Close { code, reason } => {
@@ -707,13 +707,12 @@ impl<In: DeserializeOwned, Out: Serialize, E: Encoding> TypedWebsocket<In, Out,
             match res {
                 Ok(res) => match res {
                     AxumMessage::Text(utf8_bytes) => {
-                        let e: In = E::from_bytes(utf8_bytes.into())
+                        let e: In = E::decode(utf8_bytes.into())
                             .ok_or_else(WebsocketError::deserialization)?;
                         return Ok(e);
                     }
                     AxumMessage::Binary(bytes) => {
-                        let e: In =
-                            E::from_bytes(bytes).ok_or_else(WebsocketError::deserialization)?;
+                        let e: In = E::decode(bytes).ok_or_else(WebsocketError::deserialization)?;
                         return Ok(e);
                     }