mirrors/leptos
1
0
Fork 0
mirror of https://github.com/leptos-rs/leptos.git synced 2025-12-27 09:54:41 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: prevent infinite loop when sending Result over websocket, remove Display bound (#3848)

Browse Source 
* chore: easing `Display` bound on `FromServerFnError`, #3811 follow-up

* fix: send/receive websocket data

* fix: clippy warnings

* fix: server_fn_axum example

* fix: make de/serialize_result functions public

* fix: make websocket result ser/de private

* chore: make the doc a comment and remove allow dead_code
This commit is contained in:
Greg Johnston
2025-04-23 07:52:42 -04:00
committed by GitHub
parent f83b14d76c c743f0641c
commit e65fc23fc7
8 changed files with 203 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
server_fn/src/client.rs
View File

@@ -42,7 +42,7 @@ pub trait Client<Error, InputStreamError = Error, OutputStreamError = Error> {
Output = Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Bytes> + Send + 'static,
),
Error,
>,
@@ -62,8 +62,8 @@ pub mod browser {
response::browser::BrowserResponse,
};
use bytes::Bytes;
use futures::{Sink, SinkExt, StreamExt, TryStreamExt};
use gloo_net::websocket::{events::CloseEvent, Message, WebSocketError};
use futures::{Sink, SinkExt, StreamExt};
use gloo_net::websocket::{Message, WebSocketError};
use send_wrapper::SendWrapper;
use std::future::Future;
@@ -115,7 +115,7 @@ pub mod browser {
impl futures::Stream<Item = Result<Bytes, Bytes>>
+ Send
+ 'static,
impl futures::Sink<Result<Bytes, Bytes>> + Send + 'static,
impl futures::Sink<Bytes> + Send + 'static,
),
Error,
>,
@@ -131,18 +131,19 @@ pub mod browser {
})?;
let (sink, stream) = websocket.split();
let stream = stream
.map_err(|err| {
web_sys::console::error_1(&err.to_string().into());
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Request(err.to_string()),
)
.ser()
})
.map_ok(move |msg| match msg {
let stream = stream.map(|message| match message {
Ok(message) => Ok(match message {
Message::Text(text) => Bytes::from(text),
Message::Bytes(bytes) => Bytes::from(bytes),
});
}),
Err(err) => {
web_sys::console::error_1(&err.to_string().into());
Err(OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Request(err.to_string()),
)
.ser())
}
});
let stream = SendWrapper::new(stream);
struct SendWrapperSink<S> {
@@ -195,26 +196,11 @@ pub mod browser {
}
}
let sink =
sink.with(|message: Result<Bytes, Bytes>| async move {
match message {
Ok(message) => Ok(Message::Bytes(message.into())),
Err(err) => {
let err = InputStreamError::de(err);
web_sys::console::error_1(
&js_sys::JsString::from(err.to_string()),
);
const CLOSE_CODE_ERROR: u16 = 1011;
Err(WebSocketError::ConnectionClose(
CloseEvent {
code: CLOSE_CODE_ERROR,
reason: err.to_string(),
was_clean: true,
},
))
}
}
});
let sink = sink.with(|message: Bytes| async move {
Ok::<Message, WebSocketError>(Message::Bytes(
message.into(),
))
});
let sink = SendWrapperSink::new(Box::pin(sink));
Ok((stream, sink))
@@ -243,13 +229,19 @@ pub mod reqwest {
/// Implements [`Client`] for a request made by [`reqwest`].
pub struct ReqwestClient;
impl<E: FromServerFnError + Send + 'static> Client<E> for ReqwestClient {
impl<
Error: FromServerFnError,
InputStreamError: FromServerFnError,
OutputStreamError: FromServerFnError,
> Client<Error, InputStreamError, OutputStreamError> for ReqwestClient
{
type Request = Request;
type Response = Response;
fn send(
req: Self::Request,
) -> impl Future<Output = Result<Self::Response, E>> + Send {
) -> impl Future<Output = Result<Self::Response, Error>> + Send
{
CLIENT.execute(req).map_err(|e| {
ServerFnErrorErr::Request(e.to_string()).into_app_error()
})
@@ -259,12 +251,10 @@ pub mod reqwest {
path: &str,
) -> Result<
(
impl futures::Stream<Item = Result<bytes::Bytes, Bytes>>
+ Send
+ 'static,
impl futures::Sink<Result<bytes::Bytes, Bytes>> + Send + 'static,
impl futures::Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl futures::Sink<Bytes> + Send + 'static,
),
E,
Error,
> {
let mut websocket_server_url = get_server_url().to_string();
if let Some(postfix) = websocket_server_url.strip_prefix("http://")
@@ -278,7 +268,7 @@ pub mod reqwest {
let url = format!("{}{}", websocket_server_url, path);
let (ws_stream, _) =
tokio_tungstenite::connect_async(url).await.map_err(|e| {
E::from_server_fn_error(ServerFnErrorErr::Request(
Error::from_server_fn_error(ServerFnErrorErr::Request(
e.to_string(),
))
})?;
@@ -288,25 +278,18 @@ pub mod reqwest {
Ok((
read.map(|msg| match msg {
Ok(msg) => Ok(msg.into_data()),
Err(e) => Err(E::from_server_fn_error(
Err(e) => Err(OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Request(e.to_string()),
)
.ser()),
}),
write.with(|msg: Result<Bytes, Bytes>| async move {
match msg {
Ok(msg) => {
Ok(tokio_tungstenite::tungstenite::Message::Binary(
msg,
))
}
Err(err) => {
let err = E::de(err);
Err(tokio_tungstenite::tungstenite::Error::Io(
std::io::Error::other(err.to_string()),
))
}
}
write.with(|msg: Bytes| async move {
Ok::<
tokio_tungstenite::tungstenite::Message,
tokio_tungstenite::tungstenite::Error,
>(
tokio_tungstenite::tungstenite::Message::Binary(msg)
)
}),
))
}

4
server_fn/src/error.rs
View File

@@ -561,9 +561,7 @@ impl<E: FromServerFnError> FromStr for ServerFnErrorWrapper<E> {
}
/// A trait for types that can be returned from a server function.
pub trait FromServerFnError:
std::fmt::Debug + Sized + Display + 'static
{
pub trait FromServerFnError: std::fmt::Debug + Sized + 'static {
/// The encoding strategy used to serialize and deserialize this error type. Must implement the [`Encodes`](server_fn::Encodes) trait for references to the error type.
type Encoder: Encodes<Self> + Decodes<Self>;

182
server_fn/src/lib.rs
View File

@@ -136,6 +136,7 @@ use base64::{engine::general_purpose::STANDARD_NO_PAD, DecodeError, Engine};
// re-exported to make it possible to implement a custom Client without adding a separate
// dependency on `bytes`
pub use bytes::Bytes;
use bytes::{BufMut, BytesMut};
use client::Client;
use codec::{Encoding, FromReq, FromRes, IntoReq, IntoRes};
#[doc(hidden)]
@@ -635,15 +636,19 @@ where
{
let (request_bytes, response_stream, response) =
request.try_into_websocket().await?;
let input = request_bytes.map(|request_bytes| match request_bytes {
Ok(request_bytes) => {
InputEncoding::decode(request_bytes).map_err(|e| {
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(e.to_string()),
)
})
let input = request_bytes.map(|request_bytes| {
let request_bytes = request_bytes
.map(|bytes| deserialize_result::<InputStreamError>(bytes))
.unwrap_or_else(Err);
match request_bytes {
Ok(request_bytes) => InputEncoding::decode(request_bytes)
.map_err(|e| {
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(e.to_string()),
)
}),
Err(err) => Err(InputStreamError::de(err)),
}
Err(err) => Err(InputStreamError::de(err)),
});
let boxed = Box::pin(input)
as Pin<
@@ -656,14 +661,17 @@ where
let output = server_fn(input.into()).await?;
let output = output.stream.map(|output| match output {
Ok(output) => OutputEncoding::encode(&output).map_err(|e| {
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(e.to_string()),
)
.ser()
}),
Err(err) => Err(err.ser()),
let output = output.stream.map(|output| {
let result = match output {
Ok(output) => OutputEncoding::encode(&output).map_err(|e| {
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(e.to_string()),
)
.ser()
}),
Err(err) => Err(err.ser()),
};
serialize_result(result)
});
Server::spawn(async move {
@@ -695,37 +703,42 @@ where
pin_mut!(input);
pin_mut!(sink);
while let Some(input) = input.stream.next().await {
if sink
.send(
input
.and_then(|input| {
InputEncoding::encode(&input).map_err(|e| {
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(
e.to_string(),
),
)
})
})
.map_err(|e| e.ser()),
)
.await
.is_err()
{
let result = match input {
Ok(input) => {
InputEncoding::encode(&input).map_err(|e| {
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(
e.to_string(),
),
)
.ser()
})
}
Err(err) => Err(err.ser()),
};
let result = serialize_result(result);
if sink.send(result).await.is_err() {
break;
}
}
});
// Return the output stream
let stream = stream.map(|request_bytes| match request_bytes {
Ok(request_bytes) => OutputEncoding::decode(request_bytes)
.map_err(|e| {
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(e.to_string()),
)
}),
Err(err) => Err(OutputStreamError::de(err)),
let stream = stream.map(|request_bytes| {
let request_bytes = request_bytes
.map(|bytes| deserialize_result::<OutputStreamError>(bytes))
.unwrap_or_else(Err);
match request_bytes {
Ok(request_bytes) => OutputEncoding::decode(request_bytes)
.map_err(|e| {
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(
e.to_string(),
),
)
}),
Err(err) => Err(OutputStreamError::de(err)),
}
});
let boxed = Box::pin(stream)
as Pin<
@@ -740,6 +753,51 @@ where
}
}
// Serializes a Result<Bytes, Bytes> into a single Bytes instance.
// Format: [tag: u8][content: Bytes]
// - Tag 0: Ok variant
// - Tag 1: Err variant
fn serialize_result(result: Result<Bytes, Bytes>) -> Bytes {
match result {
Ok(bytes) => {
let mut buf = BytesMut::with_capacity(1 + bytes.len());
buf.put_u8(0); // Tag for Ok variant
buf.extend_from_slice(&bytes);
buf.freeze()
}
Err(bytes) => {
let mut buf = BytesMut::with_capacity(1 + bytes.len());
buf.put_u8(1); // Tag for Err variant
buf.extend_from_slice(&bytes);
buf.freeze()
}
}
}
// Deserializes a Bytes instance back into a Result<Bytes, Bytes>.
fn deserialize_result<E: FromServerFnError>(
bytes: Bytes,
) -> Result<Bytes, Bytes> {
if bytes.is_empty() {
return Err(E::from_server_fn_error(
ServerFnErrorErr::Deserialization("Data is empty".into()),
)
.ser());
}
let tag = bytes[0];
let content = bytes.slice(1..);
match tag {
0 => Ok(content),
1 => Err(content),
_ => Err(E::from_server_fn_error(ServerFnErrorErr::Deserialization(
"Invalid data tag".into(),
))
.ser()), // Invalid tag
}
}
/// Encode format type
pub enum Format {
/// Binary representation
@@ -1218,3 +1276,45 @@ pub mod mock {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::codec::JsonEncoding;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
enum TestError {
ServerFnError(ServerFnErrorErr),
}
impl FromServerFnError for TestError {
type Encoder = JsonEncoding;
fn from_server_fn_error(value: ServerFnErrorErr) -> Self {
Self::ServerFnError(value)
}
}
#[test]
fn test_result_serialization() {
// Test Ok variant
let ok_result: Result<Bytes, Bytes> =
Ok(Bytes::from_static(b"success data"));
let serialized = serialize_result(ok_result);
let deserialized = deserialize_result::<TestError>(serialized);
assert!(deserialized.is_ok());
assert_eq!(deserialized.unwrap(), Bytes::from_static(b"success data"));
// Test Err variant
let err_result: Result<Bytes, Bytes> =
Err(Bytes::from_static(b"error details"));
let serialized = serialize_result(err_result);
let deserialized = deserialize_result::<TestError>(serialized);
assert!(deserialized.is_err());
assert_eq!(
deserialized.unwrap_err(),
Bytes::from_static(b"error details")
);
}
}

13
server_fn/src/request/actix.rs
View File

@@ -117,7 +117,7 @@ where
) -> Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl futures::Sink<Result<Bytes, Bytes>> + Send + 'static,
impl futures::Sink<Bytes> + Send + 'static,
Self::WebsocketResponse,
),
Error,
@@ -133,7 +133,7 @@ where
let (mut response_stream_tx, response_stream_rx) =
futures::channel::mpsc::channel(2048);
let (response_sink_tx, mut response_sink_rx) =
futures::channel::mpsc::channel::<Result<Bytes, Bytes>>(2048);
futures::channel::mpsc::channel::<Bytes>(2048);
actix_web::rt::spawn(async move {
loop {
@@ -142,16 +142,9 @@ where
let Some(incoming) = incoming else {
break;
};
match incoming {
Ok(message) => {
if let Err(err) = session.binary(message).await {
if let Err(err) = session.binary(incoming).await {
_ = response_stream_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Request(err.to_string())).ser()));
}
}
Err(err) => {
_ = response_stream_tx.start_send(Err(err));
}
}
},
outgoing = msg_stream.next().fuse() => {
let Some(outgoing) = outgoing else {

26
server_fn/src/request/axum.rs
View File

@@ -79,7 +79,7 @@ where
) -> Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Bytes> + Send + 'static,
Self::WebsocketResponse,
),
Error,
@@ -91,7 +91,7 @@ where
futures::stream::Once<
std::future::Ready<Result<Bytes, Bytes>>,
>,
futures::sink::Drain<Result<Bytes, Bytes>>,
futures::sink::Drain<Bytes>,
Self::WebsocketResponse,
),
Error,
@@ -117,9 +117,9 @@ where
))
})?;
let (mut outgoing_tx, outgoing_rx) =
futures::channel::mpsc::channel(2048);
let (incoming_tx, mut incoming_rx) =
futures::channel::mpsc::channel::<Result<Bytes, Bytes>>(2048);
let (incoming_tx, mut incoming_rx) =
futures::channel::mpsc::channel::<Bytes>(2048);
let response = upgrade
.on_failed_upgrade({
let mut outgoing_tx = outgoing_tx.clone();
@@ -134,18 +134,11 @@ where
let Some(incoming) = incoming else {
break;
};
match incoming {
Ok(message) => {
if let Err(err) = session.send(Message::Binary(message)).await {
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Request(err.to_string())).ser()));
}
}
Err(err) => {
_ = outgoing_tx.start_send(Err(err));
}
if let Err(err) = session.send(Message::Binary(incoming)).await {
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Request(err.to_string())).ser()));
}
},
outgoing = session.recv().fuse() => {
outgoing = session.recv().fuse() => {
let Some(outgoing) = outgoing else {
break;
};
@@ -159,6 +152,11 @@ where
Ok(Message::Text(text)) => {
_ = outgoing_tx.start_send(Ok(Bytes::from(text)));
}
Ok(Message::Ping(bytes)) => {
if session.send(Message::Pong(bytes)).await.is_err() {
break;
}
}
Ok(_other) => {}
Err(e) => {
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Response(e.to_string())).ser()));

4
server_fn/src/request/generic.rs
View File

@@ -79,7 +79,7 @@ where
) -> Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Bytes> + Send + 'static,
Self::WebsocketResponse,
),
Error,
@@ -87,7 +87,7 @@ where
Err::<
(
futures::stream::Once<std::future::Ready<Result<Bytes, Bytes>>>,
futures::sink::Drain<Result<Bytes, Bytes>>,
futures::sink::Drain<Bytes>,
Self::WebsocketResponse,
),
_,

6
server_fn/src/request/mod.rs
View File

@@ -360,7 +360,7 @@ where
Output = Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Bytes> + Send + 'static,
Self::WebsocketResponse,
),
Error,
@@ -415,7 +415,7 @@ where
) -> Result<
(
impl Stream<Item = Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Result<Bytes, Bytes>> + Send + 'static,
impl Sink<Bytes> + Send + 'static,
Self::WebsocketResponse,
),
Error,
@@ -424,7 +424,7 @@ where
Err::<
(
futures::stream::Once<std::future::Ready<Result<Bytes, Bytes>>>,
futures::sink::Drain<Result<Bytes, Bytes>>,
futures::sink::Drain<Bytes>,
Self::WebsocketResponse,
),
_,