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

feat: allow using different error types for req/resp with WebSockets, closes #3724 (#3766)

Browse Source 
* feat: allow using different error types for request/response with WebSocket

* [autofix.ci] apply automated fixes

* chore: clean up merge issues

* chore: fix custom client example

* we can't use nightly features on stable

* update flake inputs

* include gcc and glib to flake dev shell

* update expected stderr outputs server_fn/tests/invalid

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Greg Johnston <greg.johnston@gmail.com>
This commit is contained in:
Mykyta
2025-04-09 20:03:47 +03:00
committed by GitHub
parent 8bc4fd4198
commit ba7bfd8bac
15 changed files with 544 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
server_fn/src/client.rs
View File

@@ -23,16 +23,16 @@ pub fn get_server_url() -> &'static str {
/// This trait is implemented for things like a browser `fetch` request or for
/// the `reqwest` trait. It should almost never be necessary to implement it
/// yourself, unless you’re trying to use an alternative HTTP crate on the client side.
pub trait Client<E> {
pub trait Client<Error, InputStreamError = Error, OutputStreamError = Error> {
/// The type of a request sent by this client.
type Request: ClientReq<E> + Send + 'static;
type Request: ClientReq<Error> + Send + 'static;
/// The type of a response received by this client.
type Response: ClientRes<E> + Send + 'static;
type Response: ClientRes<Error> + Send + 'static;
/// Sends the request and receives a response.
fn send(
req: Self::Request,
) -> impl Future<Output = Result<Self::Response, E>> + Send;
) -> impl Future<Output = Result<Self::Response, Error>> + Send;
/// Opens a websocket connection to the server.
#[allow(clippy::type_complexity)]
@@ -41,10 +41,12 @@ pub trait Client<E> {
) -> impl Future<
Output = Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, OutputStreamError>>
+ Send
+ 'static,
impl Sink<Result<Bytes, InputStreamError>> + Send + 'static,
),
E,
Error,
>,
> + Send;
@@ -70,13 +72,19 @@ pub mod browser {
/// Implements [`Client`] for a `fetch` request in the browser.
pub struct BrowserClient;
impl<E: FromServerFnError> Client<E> for BrowserClient {
impl<
Error: FromServerFnError,
InputStreamError: FromServerFnError,
OutputStreamError: FromServerFnError,
> Client<Error, InputStreamError, OutputStreamError> for BrowserClient
{
type Request = BrowserRequest;
type Response = BrowserResponse;
fn send(
req: Self::Request,
) -> impl Future<Output = Result<Self::Response, E>> + Send {
) -> impl Future<Output = Result<Self::Response, Error>> + Send
{
SendWrapper::new(async move {
let req = req.0.take();
let RequestInner {
@@ -106,10 +114,14 @@ pub mod browser {
) -> impl Future<
Output = Result<
(
impl futures::Stream<Item = Result<Bytes, E>> + Send + 'static,
impl futures::Sink<Result<Bytes, E>> + Send + 'static,
impl futures::Stream<Item = Result<Bytes, OutputStreamError>>
+ Send
+ 'static,
impl futures::Sink<Result<Bytes, InputStreamError>>
+ Send
+ 'static,
),
E,
Error,
>,
> + Send {
SendWrapper::new(async move {
@@ -117,18 +129,18 @@ pub mod browser {
gloo_net::websocket::futures::WebSocket::open(url)
.map_err(|err| {
web_sys::console::error_1(&err.to_string().into());
E::from_server_fn_error(ServerFnErrorErr::Request(
err.to_string(),
))
Error::from_server_fn_error(
ServerFnErrorErr::Request(err.to_string()),
)
})?;
let (sink, stream) = websocket.split();
let stream = stream
.map_err(|err| {
web_sys::console::error_1(&err.to_string().into());
E::from_server_fn_error(ServerFnErrorErr::Request(
err.to_string(),
))
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Request(err.to_string()),
)
})
.map_ok(move |msg| match msg {
Message::Text(text) => Bytes::from(text),
@@ -186,22 +198,26 @@ pub mod browser {
}
}
let sink = sink.with(|message: Result<Bytes, E>| async move {
match message {
Ok(message) => Ok(Message::Bytes(message.into())),
Err(err) => {
web_sys::console::error_1(&js_sys::JsString::from(
err.ser(),
));
const CLOSE_CODE_ERROR: u16 = 1011;
Err(WebSocketError::ConnectionClose(CloseEvent {
code: CLOSE_CODE_ERROR,
reason: err.ser(),
was_clean: true,
}))
let sink = sink.with(
|message: Result<Bytes, InputStreamError>| async move {
match message {
Ok(message) => Ok(Message::Bytes(message.into())),
Err(err) => {
web_sys::console::error_1(
&js_sys::JsString::from(err.ser()),
);
const CLOSE_CODE_ERROR: u16 = 1011;
Err(WebSocketError::ConnectionClose(
CloseEvent {
code: CLOSE_CODE_ERROR,
reason: err.ser(),
was_clean: true,
},
))
}
}
}
});
},
);
let sink = SendWrapperSink::new(Box::pin(sink));
Ok((stream, sink))

179
server_fn/src/lib.rs
View File

@@ -174,9 +174,13 @@ pub use xxhash_rust;
type ServerFnServerRequest<Fn> = <<Fn as ServerFn>::Server as crate::Server<
<Fn as ServerFn>::Error,
<Fn as ServerFn>::InputStreamError,
<Fn as ServerFn>::OutputStreamError,
>>::Request;
type ServerFnServerResponse<Fn> = <<Fn as ServerFn>::Server as crate::Server<
<Fn as ServerFn>::Error,
<Fn as ServerFn>::InputStreamError,
<Fn as ServerFn>::OutputStreamError,
>>::Response;
/// Defines a function that runs only on the server, but can be called from the server or the client.
@@ -215,12 +219,20 @@ pub trait ServerFn: Send + Sized {
/// The type of the HTTP client that will send the request from the client side.
///
/// For example, this might be `gloo-net` in the browser, or `reqwest` for a desktop app.
type Client: Client<Self::Error>;
type Client: Client<
Self::Error,
Self::InputStreamError,
Self::OutputStreamError,
>;
/// The type of the HTTP server that will send the response from the server side.
///
/// For example, this might be `axum` or `actix-web`.
type Server: Server<Self::Error>;
type Server: Server<
Self::Error,
Self::InputStreamError,
Self::OutputStreamError,
>;
/// The protocol the server function uses to communicate with the client.
type Protocol: Protocol<
@@ -229,6 +241,8 @@ pub trait ServerFn: Send + Sized {
Self::Client,
Self::Server,
Self::Error,
Self::InputStreamError,
Self::OutputStreamError,
>;
/// The return type of the server function.
@@ -237,8 +251,15 @@ pub trait ServerFn: Send + Sized {
/// *from* `ClientResponse` when received by the client.
type Output: Send;
/// The type of the error on the server function. Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
/// The type of error in the server function return.
/// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
type Error: FromServerFnError + Send + Sync;
/// The type of error in the server function for stream items sent from the client to the server.
/// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
type InputStreamError: FromServerFnError + Send + Sync;
/// The type of error in the server function for stream items sent from the server to the client.
/// Typically [`ServerFnError`], but allowed to be any type that implements [`FromServerFnError`].
type OutputStreamError: FromServerFnError + Send + Sync;
/// Returns [`Self::PATH`].
fn url() -> &'static str {
@@ -288,6 +309,8 @@ pub trait ServerFn: Send + Sized {
(
<<Self as ServerFn>::Server as crate::Server<
Self::Error,
Self::InputStreamError,
Self::OutputStreamError,
>>::Response::error_response(
Self::PATH, e.ser()
),
@@ -331,10 +354,17 @@ pub trait ServerFn: Send + Sized {
/// The protocol that a server function uses to communicate with the client. This trait handles
/// the server and client side of running a server function. It is implemented for the [`Http`] and
/// [`Websocket`] protocols and can be used to implement custom protocols.
pub trait Protocol<Input, Output, Client, Server, E>
where
Server: crate::Server<E>,
Client: crate::Client<E>,
pub trait Protocol<
Input,
Output,
Client,
Server,
Error,
InputStreamError = Error,
OutputStreamError = Error,
> where
Server: crate::Server<Error, InputStreamError, OutputStreamError>,
Client: crate::Client<Error, InputStreamError, OutputStreamError>,
{
/// The HTTP method used for requests.
const METHOD: Method;
@@ -344,17 +374,17 @@ where
fn run_server<F, Fut>(
request: Server::Request,
server_fn: F,
) -> impl Future<Output = Result<Server::Response, E>> + Send
) -> impl Future<Output = Result<Server::Response, Error>> + Send
where
F: Fn(Input) -> Fut + Send,
Fut: Future<Output = Result<Output, E>> + Send;
Fut: Future<Output = Result<Output, Error>> + Send;
/// Run the server function on the client. The implementation should handle serializing the
/// input, sending the request, and deserializing the output.
fn run_client(
path: &str,
input: Input,
) -> impl Future<Output = Result<Output, E>> + Send;
) -> impl Future<Output = Result<Output, Error>> + Send;
}
/// The http protocol with specific input and output encodings for the request and response. This is
@@ -561,18 +591,30 @@ impl<
OutputEncoding,
Client,
Server,
E,
> Protocol<Input, BoxedStream<OutputItem, E>, Client, Server, E>
for Websocket<InputEncoding, OutputEncoding>
Error,
InputStreamError,
OutputStreamError,
>
Protocol<
Input,
BoxedStream<OutputItem, OutputStreamError>,
Client,
Server,
Error,
InputStreamError,
OutputStreamError,
> for Websocket<InputEncoding, OutputEncoding>
where
Input: Deref<Target = BoxedStream<InputItem, E>>
+ Into<BoxedStream<InputItem, E>>
+ From<BoxedStream<InputItem, E>>,
Input: Deref<Target = BoxedStream<InputItem, InputStreamError>>
+ Into<BoxedStream<InputItem, InputStreamError>>
+ From<BoxedStream<InputItem, InputStreamError>>,
InputEncoding: Encodes<InputItem> + Decodes<InputItem>,
OutputEncoding: Encodes<OutputItem> + Decodes<OutputItem>,
Server: crate::Server<E>,
E: FromServerFnError + Send,
Client: crate::Client<E>,
InputStreamError: FromServerFnError + Send,
OutputStreamError: FromServerFnError + Send,
Error: FromServerFnError + Send,
Server: crate::Server<Error, InputStreamError, OutputStreamError>,
Client: crate::Client<Error, InputStreamError, OutputStreamError>,
OutputItem: Send + 'static,
InputItem: Send + 'static,
{
@@ -581,34 +623,44 @@ where
async fn run_server<F, Fut>(
request: Server::Request,
server_fn: F,
) -> Result<Server::Response, E>
) -> Result<Server::Response, Error>
where
F: Fn(Input) -> Fut + Send,
Fut: Future<Output = Result<BoxedStream<OutputItem, E>, E>> + Send,
Fut: Future<
Output = Result<
BoxedStream<OutputItem, OutputStreamError>,
Error,
>,
> + Send,
{
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| {
E::from_server_fn_error(ServerFnErrorErr::Deserialization(
e.to_string(),
))
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(e.to_string()),
)
})
}
Err(err) => Err(err),
});
let boxed = Box::pin(input)
as Pin<Box<dyn Stream<Item = Result<InputItem, E>> + Send>>;
as Pin<
Box<
dyn Stream<Item = Result<InputItem, InputStreamError>>
+ Send,
>,
>;
let input = BoxedStream { stream: boxed };
let output = server_fn(input.into()).await?;
let output = output.stream.map(|output| match output {
Ok(output) => OutputEncoding::encode(output).map_err(|e| {
E::from_server_fn_error(ServerFnErrorErr::Serialization(
e.to_string(),
))
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(e.to_string()),
)
}),
Err(err) => Err(err),
});
@@ -629,8 +681,9 @@ where
fn run_client(
path: &str,
input: Input,
) -> impl Future<Output = Result<BoxedStream<OutputItem, E>, E>> + Send
{
) -> impl Future<
Output = Result<BoxedStream<OutputItem, OutputStreamError>, Error>,
> + Send {
let input = input.into();
async move {
@@ -644,7 +697,7 @@ where
if sink
.send(input.and_then(|input| {
InputEncoding::encode(input).map_err(|e| {
E::from_server_fn_error(
InputStreamError::from_server_fn_error(
ServerFnErrorErr::Serialization(
e.to_string(),
),
@@ -663,14 +716,19 @@ where
let stream = stream.map(|request_bytes| match request_bytes {
Ok(request_bytes) => OutputEncoding::decode(request_bytes)
.map_err(|e| {
E::from_server_fn_error(
OutputStreamError::from_server_fn_error(
ServerFnErrorErr::Deserialization(e.to_string()),
)
}),
Err(err) => Err(err),
});
let boxed = Box::pin(stream)
as Pin<Box<dyn Stream<Item = Result<OutputItem, E>> + Send>>;
as Pin<
Box<
dyn Stream<Item = Result<OutputItem, OutputStreamError>>
+ Send,
>,
>;
let output = BoxedStream { stream: boxed };
Ok(output)
}
@@ -738,13 +796,25 @@ impl<Req, Res> ServerFnTraitObj<Req, Res> {
/// Converts the relevant parts of a server function into a trait object.
pub const fn new<
S: ServerFn<
Server: crate::Server<S::Error, Request = Req, Response = Res>,
Server: crate::Server<
S::Error,
S::InputStreamError,
S::OutputStreamError,
Request = Req,
Response = Res,
>,
>,
>(
handler: fn(Req) -> Pin<Box<dyn Future<Output = Res> + Send>>,
) -> Self
where
Req: crate::Req<S::Error, WebsocketResponse = Res> + Send + 'static,
Req: crate::Req<
S::Error,
S::InputStreamError,
S::OutputStreamError,
WebsocketResponse = Res,
> + Send
+ 'static,
Res: crate::TryRes<S::Error> + Send + 'static,
{
Self {
@@ -848,14 +918,21 @@ pub mod axum {
/// The axum server function backend
pub struct AxumServerFnBackend;
impl<E: FromServerFnError + Send + Sync> Server<E> for AxumServerFnBackend {
impl<Error, InputStreamError, OutputStreamError>
Server<Error, InputStreamError, OutputStreamError>
for AxumServerFnBackend
where
Error: FromServerFnError + Send + Sync,
InputStreamError: FromServerFnError + Send + Sync,
OutputStreamError: FromServerFnError + Send + Sync,
{
type Request = Request<Body>;
type Response = Response<Body>;
#[allow(unused_variables)]
fn spawn(
future: impl Future<Output = ()> + Send + 'static,
) -> Result<(), E> {
) -> Result<(), Error> {
#[cfg(feature = "axum")]
{
tokio::spawn(future);
@@ -863,7 +940,7 @@ pub mod axum {
}
#[cfg(not(feature = "axum"))]
{
Err(E::from_server_fn_error(
Err(Error::from_server_fn_error(
crate::error::ServerFnErrorErr::Request(
"No async runtime available. You need to either \
enable the full axum feature to pull in tokio, or \
@@ -884,6 +961,8 @@ pub mod axum {
T: ServerFn<
Server: crate::Server<
T::Error,
T::InputStreamError,
T::OutputStreamError,
Request = Request<Body>,
Response = Response<Body>,
>,
@@ -966,13 +1045,20 @@ pub mod actix {
/// The actix server function backend
pub struct ActixServerFnBackend;
impl<E: FromServerFnError + Send + Sync> Server<E> for ActixServerFnBackend {
impl<Error, InputStreamError, OutputStreamError>
Server<Error, InputStreamError, OutputStreamError>
for ActixServerFnBackend
where
Error: FromServerFnError + Send + Sync,
InputStreamError: FromServerFnError + Send + Sync,
OutputStreamError: FromServerFnError + Send + Sync,
{
type Request = ActixRequest;
type Response = ActixResponse;
fn spawn(
future: impl Future<Output = ()> + Send + 'static,
) -> Result<(), E> {
) -> Result<(), Error> {
actix_web::rt::spawn(future);
Ok(())
}
@@ -986,6 +1072,8 @@ pub mod actix {
T: ServerFn<
Server: crate::Server<
T::Error,
T::InputStreamError,
T::OutputStreamError,
Request = ActixRequest,
Response = ActixResponse,
>,
@@ -1075,13 +1163,20 @@ pub mod mock {
/// server type when compiling for the client.
pub struct BrowserMockServer;
impl<E: Send + 'static> crate::server::Server<E> for BrowserMockServer {
impl<Error, InputStreamError, OutputStreamError>
crate::server::Server<Error, InputStreamError, OutputStreamError>
for BrowserMockServer
where
Error: Send + 'static,
InputStreamError: Send + 'static,
OutputStreamError: Send + 'static,
{
type Request = crate::request::BrowserMockReq;
type Response = crate::response::BrowserMockRes;
fn spawn(
_: impl Future<Output = ()> + Send + 'static,
) -> Result<(), E> {
) -> Result<(), Error> {
unreachable!()
}
}

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

@@ -38,9 +38,12 @@ impl From<(HttpRequest, Payload)> for ActixRequest {
}
}
impl<E> Req<E> for ActixRequest
impl<Error, InputStreamError, OutputStreamError>
Req<Error, InputStreamError, OutputStreamError> for ActixRequest
where
E: FromServerFnError + Send,
Error: FromServerFnError + Send,
InputStreamError: FromServerFnError + Send,
OutputStreamError: FromServerFnError + Send,
{
type WebsocketResponse = ActixResponse;
@@ -60,7 +63,9 @@ where
self.header("Referer")
}
fn try_into_bytes(self) -> impl Future<Output = Result<Bytes, E>> + Send {
fn try_into_bytes(
self,
) -> impl Future<Output = Result<Bytes, Error>> + Send {
// Actix is going to keep this on a single thread anyway so it's fine to wrap it
// with SendWrapper, which makes it `Send` but will panic if it moves to another thread
SendWrapper::new(async move {
@@ -72,18 +77,20 @@ where
})
}
fn try_into_string(self) -> impl Future<Output = Result<String, E>> + Send {
fn try_into_string(
self,
) -> impl Future<Output = Result<String, Error>> + Send {
// Actix is going to keep this on a single thread anyway so it's fine to wrap it
// with SendWrapper, which makes it `Send` but will panic if it moves to another thread
SendWrapper::new(async move {
let payload = self.0.take().1;
let bytes = payload.to_bytes().await.map_err(|e| {
E::from_server_fn_error(ServerFnErrorErr::Deserialization(
Error::from_server_fn_error(ServerFnErrorErr::Deserialization(
e.to_string(),
))
})?;
String::from_utf8(bytes.into()).map_err(|e| {
E::from_server_fn_error(ServerFnErrorErr::Deserialization(
Error::from_server_fn_error(ServerFnErrorErr::Deserialization(
e.to_string(),
))
})
@@ -92,7 +99,7 @@ where
fn try_into_stream(
self,
) -> Result<impl Stream<Item = Result<Bytes, E>> + Send, E> {
) -> Result<impl Stream<Item = Result<Bytes, Error>> + Send, Error> {
let payload = self.0.take().1;
let stream = payload.map(|res| {
res.map_err(|e| {
@@ -107,16 +114,16 @@ where
self,
) -> Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl futures::Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, InputStreamError>> + Send + 'static,
impl futures::Sink<Result<Bytes, OutputStreamError>> + Send + 'static,
Self::WebsocketResponse,
),
E,
Error,
> {
let (request, payload) = self.0.take();
let (response, mut session, mut msg_stream) =
actix_ws::handle(&request, payload).map_err(|e| {
E::from_server_fn_error(ServerFnErrorErr::Request(
Error::from_server_fn_error(ServerFnErrorErr::Request(
e.to_string(),
))
})?;
@@ -124,7 +131,9 @@ 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(2048);
futures::channel::mpsc::channel::<Result<Bytes, OutputStreamError>>(
2048,
);
actix_web::rt::spawn(async move {
loop {
@@ -136,11 +145,11 @@ where
match incoming {
Ok(message) => {
if let Err(err) = session.binary(message).await {
_ = response_stream_tx.start_send(Err(E::from_server_fn_error(ServerFnErrorErr::Request(err.to_string()))));
_ = response_stream_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Request(err.to_string()))));
}
}
Err(err) => {
_ = response_stream_tx.start_send(Err(err));
_ = response_stream_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::ServerError(err.ser()))));
}
}
},
@@ -166,7 +175,7 @@ where
Ok(_other) => {
}
Err(e) => {
_ = response_stream_tx.start_send(Err(E::from_server_fn_error(ServerFnErrorErr::Response(e.to_string()))));
_ = response_stream_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Response(e.to_string()))));
}
}
}

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

@@ -14,9 +14,12 @@ use http::{
use http_body_util::BodyExt;
use std::borrow::Cow;
impl<E> Req<E> for Request<Body>
impl<Error, InputStreamError, OutputStreamError>
Req<Error, InputStreamError, OutputStreamError> for Request<Body>
where
E: FromServerFnError + Send,
Error: FromServerFnError + Send,
InputStreamError: FromServerFnError + Send,
OutputStreamError: FromServerFnError + Send,
{
type WebsocketResponse = Response;
@@ -42,7 +45,7 @@ where
.map(|h| String::from_utf8_lossy(h.as_bytes()))
}
async fn try_into_bytes(self) -> Result<Bytes, E> {
async fn try_into_bytes(self) -> Result<Bytes, Error> {
let (_parts, body) = self.into_parts();
body.collect().await.map(|c| c.to_bytes()).map_err(|e| {
@@ -50,8 +53,8 @@ where
})
}
async fn try_into_string(self) -> Result<String, E> {
let bytes = self.try_into_bytes().await?;
async fn try_into_string(self) -> Result<String, Error> {
let bytes = Req::<Error>::try_into_bytes(self).await?;
String::from_utf8(bytes.to_vec()).map_err(|e| {
ServerFnErrorErr::Deserialization(e.to_string()).into_app_error()
})
@@ -59,7 +62,8 @@ where
fn try_into_stream(
self,
) -> Result<impl Stream<Item = Result<Bytes, E>> + Send + 'static, E> {
) -> Result<impl Stream<Item = Result<Bytes, Error>> + Send + 'static, Error>
{
Ok(self.into_body().into_data_stream().map(|chunk| {
chunk.map_err(|e| {
ServerFnErrorErr::Deserialization(e.to_string())
@@ -72,22 +76,24 @@ where
self,
) -> Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, InputStreamError>> + Send + 'static,
impl Sink<Result<Bytes, OutputStreamError>> + Send + 'static,
Self::WebsocketResponse,
),
E,
Error,
> {
#[cfg(not(feature = "axum"))]
{
Err::<
(
futures::stream::Once<std::future::Ready<Result<Bytes, E>>>,
futures::sink::Drain<Result<Bytes, E>>,
futures::stream::Once<
std::future::Ready<Result<Bytes, InputStreamError>>,
>,
futures::sink::Drain<Result<Bytes, OutputStreamError>>,
Self::WebsocketResponse,
),
_,
>(E::from_server_fn_error(
Error,
>(Error::from_server_fn_error(
crate::ServerFnErrorErr::Response(
"Websocket connections not supported for Axum when the \
`axum` feature is not enabled on the `server_fn` crate."
@@ -104,19 +110,21 @@ where
axum::extract::ws::WebSocketUpgrade::from_request(self, &())
.await
.map_err(|err| {
E::from_server_fn_error(ServerFnErrorErr::Request(
Error::from_server_fn_error(ServerFnErrorErr::Request(
err.to_string(),
))
})?;
let (mut outgoing_tx, outgoing_rx) =
futures::channel::mpsc::channel(2048);
let (incoming_tx, mut incoming_rx) =
futures::channel::mpsc::channel::<Result<Bytes, E>>(2048);
futures::channel::mpsc::channel::<
Result<Bytes, OutputStreamError>,
>(2048);
let response = upgrade
.on_failed_upgrade({
let mut outgoing_tx = outgoing_tx.clone();
move |err: axum::Error| {
_ = outgoing_tx.start_send(Err(E::from_server_fn_error(ServerFnErrorErr::Response(err.to_string()))));
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Response(err.to_string()))));
}
})
.on_upgrade(|mut session| async move {
@@ -129,11 +137,11 @@ where
match incoming {
Ok(message) => {
if let Err(err) = session.send(Message::Binary(message)).await {
_ = outgoing_tx.start_send(Err(E::from_server_fn_error(ServerFnErrorErr::Request(err.to_string()))));
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Request(err.to_string()))));
}
}
Err(err) => {
_ = outgoing_tx.start_send(Err(err));
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::ServerError(err.ser()))));
}
}
},
@@ -153,7 +161,7 @@ where
}
Ok(_other) => {}
Err(e) => {
_ = outgoing_tx.start_send(Err(E::from_server_fn_error(ServerFnErrorErr::Response(e.to_string()))));
_ = outgoing_tx.start_send(Err(InputStreamError::from_server_fn_error(ServerFnErrorErr::Response(e.to_string()))));
}
}
}

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

@@ -24,17 +24,20 @@ use futures::{
use http::{Request, Response};
use std::borrow::Cow;
impl<E> Req<E> for Request<Bytes>
impl<Error, InputStreamError, OutputStreamError>
Req<Error, InputStreamError, OutputStreamError> for Request<Bytes>
where
E: FromServerFnError + Send,
Error: FromServerFnError + Send,
InputStreamError: FromServerFnError + Send,
OutputStreamError: FromServerFnError + Send,
{
type WebsocketResponse = Response<Bytes>;
async fn try_into_bytes(self) -> Result<Bytes, E> {
async fn try_into_bytes(self) -> Result<Bytes, Error> {
Ok(self.into_body())
}
async fn try_into_string(self) -> Result<String, E> {
async fn try_into_string(self) -> Result<String, Error> {
String::from_utf8(self.into_body().into()).map_err(|err| {
ServerFnErrorErr::Deserialization(err.to_string()).into_app_error()
})
@@ -42,7 +45,8 @@ where
fn try_into_stream(
self,
) -> Result<impl Stream<Item = Result<Bytes, E>> + Send + 'static, E> {
) -> Result<impl Stream<Item = Result<Bytes, Error>> + Send + 'static, Error>
{
Ok(stream::iter(self.into_body())
.ready_chunks(16)
.map(|chunk| Ok(Bytes::from(chunk))))
@@ -74,21 +78,25 @@ where
self,
) -> Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, InputStreamError>> + Send + 'static,
impl Sink<Result<Bytes, OutputStreamError>> + Send + 'static,
Self::WebsocketResponse,
),
E,
Error,
> {
Err::<
(
futures::stream::Once<std::future::Ready<Result<Bytes, E>>>,
futures::sink::Drain<Result<Bytes, E>>,
futures::stream::Once<
std::future::Ready<Result<Bytes, InputStreamError>>,
>,
futures::sink::Drain<Result<Bytes, OutputStreamError>>,
Self::WebsocketResponse,
),
_,
>(E::from_server_fn_error(crate::ServerFnErrorErr::Response(
"Websockets are not supported on this platform.".to_string(),
)))
>(Error::from_server_fn_error(
crate::ServerFnErrorErr::Response(
"Websockets are not supported on this platform.".to_string(),
),
))
}
}

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

@@ -318,7 +318,7 @@ where
}
/// Represents the request as received by the server.
pub trait Req<E>
pub trait Req<Error, InputStreamError = Error, OutputStreamError = Error>
where
Self: Sized,
{
@@ -338,15 +338,19 @@ where
fn referer(&self) -> Option<Cow<'_, str>>;
/// Attempts to extract the body of the request into [`Bytes`].
fn try_into_bytes(self) -> impl Future<Output = Result<Bytes, E>> + Send;
fn try_into_bytes(
self,
) -> impl Future<Output = Result<Bytes, Error>> + Send;
/// Attempts to convert the body of the request into a string.
fn try_into_string(self) -> impl Future<Output = Result<String, E>> + Send;
fn try_into_string(
self,
) -> impl Future<Output = Result<String, Error>> + Send;
/// Attempts to convert the body of the request into a stream of bytes.
fn try_into_stream(
self,
) -> Result<impl Stream<Item = Result<Bytes, E>> + Send + 'static, E>;
) -> Result<impl Stream<Item = Result<Bytes, Error>> + Send + 'static, Error>;
/// Attempts to convert the body of the request into a websocket handle.
#[allow(clippy::type_complexity)]
@@ -355,11 +359,11 @@ where
) -> impl Future<
Output = Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, InputStreamError>> + Send + 'static,
impl Sink<Result<Bytes, OutputStreamError>> + Send + 'static,
Self::WebsocketResponse,
),
E,
Error,
>,
> + Send;
}
@@ -368,7 +372,13 @@ where
/// when compiling for the browser.
pub struct BrowserMockReq;
impl<E: Send + 'static> Req<E> for BrowserMockReq {
impl<Error, InputStreamError, OutputStreamError>
Req<Error, InputStreamError, OutputStreamError> for BrowserMockReq
where
Error: Send + 'static,
InputStreamError: Send + 'static,
OutputStreamError: Send + 'static,
{
type WebsocketResponse = crate::response::BrowserMockRes;
fn as_query(&self) -> Option<&str> {
@@ -386,17 +396,17 @@ impl<E: Send + 'static> Req<E> for BrowserMockReq {
fn referer(&self) -> Option<Cow<'_, str>> {
unreachable!()
}
async fn try_into_bytes(self) -> Result<Bytes, E> {
async fn try_into_bytes(self) -> Result<Bytes, Error> {
unreachable!()
}
async fn try_into_string(self) -> Result<String, E> {
async fn try_into_string(self) -> Result<String, Error> {
unreachable!()
}
fn try_into_stream(
self,
) -> Result<impl Stream<Item = Result<Bytes, E>> + Send, E> {
) -> Result<impl Stream<Item = Result<Bytes, Error>> + Send, Error> {
Ok(futures::stream::once(async { unreachable!() }))
}
@@ -404,17 +414,19 @@ impl<E: Send + 'static> Req<E> for BrowserMockReq {
self,
) -> Result<
(
impl Stream<Item = Result<Bytes, E>> + Send + 'static,
impl Sink<Result<Bytes, E>> + Send + 'static,
impl Stream<Item = Result<Bytes, InputStreamError>> + Send + 'static,
impl Sink<Result<Bytes, OutputStreamError>> + Send + 'static,
Self::WebsocketResponse,
),
E,
Error,
> {
#[allow(unreachable_code)]
Err::<
(
futures::stream::Once<std::future::Ready<Result<Bytes, E>>>,
futures::sink::Drain<Result<Bytes, E>>,
futures::stream::Once<
std::future::Ready<Result<Bytes, InputStreamError>>,
>,
futures::sink::Drain<Result<Bytes, OutputStreamError>>,
Self::WebsocketResponse,
),
_,

14
server_fn/src/server.rs
View File

@@ -10,15 +10,21 @@ use std::future::Future;
/// This trait is implemented for any server backend for server functions including
/// `axum` and `actix-web`. It should almost never be necessary to implement it
/// yourself, unless you’re trying to use an alternative HTTP server.
pub trait Server<E> {
pub trait Server<Error, InputStreamError = Error, OutputStreamError = Error> {
/// The type of the HTTP request when received by the server function on the server side.
type Request: Req<E, WebsocketResponse = Self::Response> + Send + 'static;
type Request: Req<
Error,
InputStreamError,
OutputStreamError,
WebsocketResponse = Self::Response,
> + Send
+ 'static;
/// The type of the HTTP response returned by the server function on the server side.
type Response: Res + TryRes<E> + Send + 'static;
type Response: Res + TryRes<Error> + Send + 'static;
/// Spawn an async task on the server.
fn spawn(
future: impl Future<Output = ()> + Send + 'static,
) -> Result<(), E>;
) -> Result<(), Error>;
}

41
server_fn/tests/invalid/aliased_return_full.stderr
View File

@@ -9,8 +9,13 @@ error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
note: required by a bound in `server_fn::ServerFn::Client`
--> src/lib.rs
|
| type Client: Client<Self::Error>;
| ^^^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Client`
| type Client: Client<
| __________________^
| | Self::Error,
| | Self::InputStreamError,
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Client`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
@@ -30,8 +35,8 @@ note: required by a bound in `server_fn::ServerFn::Protocol`
| | Self,
| | Self::Output,
| | Self::Client,
| | Self::Server,
| | Self::Error,
... |
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Protocol`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
@@ -49,3 +54,31 @@ note: required by a bound in `server_fn::ServerFn::Error`
| type Error: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Error`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_full.rs:11:1
|
11 | #[server]
| ^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::InputStreamError`
--> src/lib.rs
|
| type InputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::InputStreamError`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_full.rs:11:1
|
11 | #[server]
| ^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::OutputStreamError`
--> src/lib.rs
|
| type OutputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::OutputStreamError`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)

39
server_fn/tests/invalid/aliased_return_none.stderr
View File

@@ -9,8 +9,13 @@ error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
note: required by a bound in `server_fn::ServerFn::Client`
--> src/lib.rs
|
| type Client: Client<Self::Error>;
| ^^^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Client`
| type Client: Client<
| __________________^
| | Self::Error,
| | Self::InputStreamError,
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Client`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
@@ -30,8 +35,8 @@ note: required by a bound in `server_fn::ServerFn::Protocol`
| | Self,
| | Self::Output,
| | Self::Client,
| | Self::Server,
| | Self::Error,
... |
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Protocol`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
@@ -48,3 +53,29 @@ note: required by a bound in `server_fn::ServerFn::Error`
|
| type Error: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Error`
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_none.rs:10:50
|
10 | pub async fn no_alias_result() -> Result<String, InvalidError> {
| ^^^^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::InputStreamError`
--> src/lib.rs
|
| type InputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::InputStreamError`
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_none.rs:10:50
|
10 | pub async fn no_alias_result() -> Result<String, InvalidError> {
| ^^^^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::OutputStreamError`
--> src/lib.rs
|
| type OutputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::OutputStreamError`

41
server_fn/tests/invalid/aliased_return_part.stderr
View File

@@ -9,8 +9,13 @@ error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
note: required by a bound in `server_fn::ServerFn::Client`
--> src/lib.rs
|
| type Client: Client<Self::Error>;
| ^^^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Client`
| type Client: Client<
| __________________^
| | Self::Error,
| | Self::InputStreamError,
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Client`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
@@ -30,8 +35,8 @@ note: required by a bound in `server_fn::ServerFn::Protocol`
| | Self,
| | Self::Output,
| | Self::Client,
| | Self::Server,
| | Self::Error,
... |
| | Self::OutputStreamError,
| | >;
| |_____^ required by this bound in `ServerFn::Protocol`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
@@ -49,3 +54,31 @@ note: required by a bound in `server_fn::ServerFn::Error`
| type Error: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::Error`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_part.rs:11:1
|
11 | #[server]
| ^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::InputStreamError`
--> src/lib.rs
|
| type InputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::InputStreamError`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)
error[E0277]: the trait bound `InvalidError: FromServerFnError` is not satisfied
--> tests/invalid/aliased_return_part.rs:11:1
|
11 | #[server]
| ^^^^^^^^^ the trait `FromServerFnError` is not implemented for `InvalidError`
|
= help: the trait `FromServerFnError` is implemented for `ServerFnError<CustErr>`
note: required by a bound in `server_fn::ServerFn::OutputStreamError`
--> src/lib.rs
|
| type OutputStreamError: FromServerFnError + Send + Sync;
| ^^^^^^^^^^^^^^^^^ required by this bound in `ServerFn::OutputStreamError`
= note: this error originates in the attribute macro `server` (in Nightly builds, run with -Z macro-backtrace for more info)