Add warp adapter

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "packages/server/server-macro",
     "packages/server/examples/axum-hello-world",
     "packages/server/examples/salvo-hello-world",
+    "packages/server/examples/warp-hello-world",
     "docs/guide",
 ]
 

packages/server/Cargo.toml

@@ -11,6 +11,7 @@ server_macro = { path = "server-macro" }
 
 # warp
 warp = { version = "0.3.3", optional = true }
+http-body = { version = "0.4.5", optional = true }
 
 # axum
 axum = { version = "0.6.1", optional = true }
@@ -31,7 +32,7 @@ tokio = { version = "1.27.0", features = ["full"], optional = true }
 
 [features]
 default = []
-warp = ["dep:warp"]
+warp = ["dep:warp", "http-body"]
 axum = ["dep:axum", "tower-http", "hyper"]
 salvo = ["dep:salvo", "hyper"]
 ssr = ["server_fn/ssr", "tokio", "dioxus-ssr"]

packages/server/examples/warp-hello-world/.gitignore

@@ -0,0 +1,2 @@
+dist
+target

packages/server/examples/warp-hello-world/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "warp-hello-world"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+dioxus-web = { path = "../../../web", features=["hydrate"], optional = true }
+dioxus = { path = "../../../dioxus" }
+dioxus-server = { path = "../../" }
+tokio = { version = "1.27.0", features = ["full"], optional = true }
+serde = "1.0.159"
+tracing-subscriber = "0.3.16"
+tracing = "0.1.37"
+warp = { version = "0.3.3", optional = true }
+
+[features]
+ssr = ["warp", "tokio", "dioxus-server/ssr", "dioxus-server/warp"]
+web = ["dioxus-web"]

packages/server/examples/warp-hello-world/src/main.rs

@@ -0,0 +1,65 @@
+//! Run with:
+//!
+//! ```sh
+//! dioxus build --features web
+//! cargo run --features ssr
+//! ```
+
+#![allow(non_snake_case)]
+use dioxus::prelude::*;
+use dioxus_server::prelude::*;
+
+fn main() {
+    #[cfg(feature = "web")]
+    dioxus_web::launch_cfg(app, dioxus_web::Config::new().hydrate(true));
+    #[cfg(feature = "ssr")]
+    {
+        PostServerData::register().unwrap();
+        GetServerData::register().unwrap();
+        tokio::runtime::Runtime::new()
+            .unwrap()
+            .block_on(async move {
+                let routes = serve_dioxus_application(
+                    ServeConfig::new(app, ()).head(r#"<title>Hello World!</title>"#),
+                );
+                warp::serve(routes).run(([127, 0, 0, 1], 8080)).await;
+            });
+    }
+}
+
+fn app(cx: Scope) -> Element {
+    let mut count = use_state(cx, || 0);
+    let text = use_state(cx, || "...".to_string());
+
+    cx.render(rsx! {
+        h1 { "High-Five counter: {count}" }
+        button { onclick: move |_| count += 1, "Up high!" }
+        button { onclick: move |_| count -= 1, "Down low!" }
+        button {
+            onclick: move |_| {
+                to_owned![text];
+                async move {
+                    if let Ok(data) = get_server_data().await {
+                        println!("Client received: {}", data);
+                        text.set(data.clone());
+                        post_server_data(data).await.unwrap();
+                    }
+                }
+            },
+            "Run a server function"
+        }
+        "Server said: {text}"
+    })
+}
+
+#[server(PostServerData)]
+async fn post_server_data(data: String) -> Result<(), ServerFnError> {
+    println!("Server received: {}", data);
+
+    Ok(())
+}
+
+#[server(GetServerData)]
+async fn get_server_data() -> Result<String, ServerFnError> {
+    Ok("Hello from the server!".to_string())
+}

packages/server/src/adapters/mod.rs

@@ -2,3 +2,5 @@
 pub mod axum_adapter;
 #[cfg(feature = "salvo")]
 pub mod salvo_adapter;
+#[cfg(feature = "warp")]
+pub mod warp_adapter;

packages/server/src/adapters/salvo_adapter.rs

@@ -132,12 +132,12 @@ impl ServerFnHandler {
                                     x-www-form-urlencoded",
                             ),
                         );
-                        res.render(data);
+                        res.write_body(data).unwrap();
                     }
                     Payload::Json(data) => {
                         res.headers_mut()
                             .insert("Content-Type", HeaderValue::from_static("application/json"));
-                        res.render(data);
+                        res.write_body(data).unwrap();
                     }
                 }
             }

packages/server/src/adapters/warp_adapter.rs

@@ -1,28 +1,136 @@
-use crate::{LiveViewError, LiveViewSocket};
-use futures_util::{SinkExt, StreamExt};
-use warp::ws::{Message, WebSocket};
+use std::{error::Error, sync::Arc};
 
-/// Convert a warp websocket into a LiveViewSocket
-///
-/// This is required to launch a LiveView app using the warp web framework
-pub fn warp_socket(ws: WebSocket) -> impl LiveViewSocket {
-    ws.map(transform_rx)
-        .with(transform_tx)
-        .sink_map_err(|_| LiveViewError::SendingFailed)
+use server_fn::{Payload, ServerFunctionRegistry};
+use tokio::task::spawn_blocking;
+use warp::{
+    filters::BoxedFilter,
+    http::{Response, StatusCode},
+    hyper::{body::Bytes, HeaderMap},
+    path, Filter, Reply,
+};
+
+use crate::{
+    dioxus_ssr_html,
+    serve::ServeConfig,
+    server_fn::{DioxusServerContext, DioxusServerFnRegistry, ServerFnTraitObj},
+};
+
+pub fn register_server_fns(server_fn_route: &'static str) -> BoxedFilter<(impl Reply,)> {
+    let mut filter: Option<BoxedFilter<(_,)>> = None;
+    for server_fn_path in DioxusServerFnRegistry::paths_registered() {
+        let func = DioxusServerFnRegistry::get(server_fn_path).unwrap();
+        let full_route = format!("{server_fn_route}/{server_fn_path}")
+            .trim_start_matches('/')
+            .to_string();
+        let route = path(full_route)
+            .and(warp::post())
+            .and(warp::header::headers_cloned())
+            .and(warp::body::bytes())
+            .and_then(move |headers: HeaderMap, body| {
+                let func = func.clone();
+                async move { server_fn_handler(DioxusServerContext {}, func, headers, body).await }
+            })
+            .boxed();
+        if let Some(boxed_filter) = filter.take() {
+            filter = Some(boxed_filter.or(route).unify().boxed());
+        } else {
+            filter = Some(route.boxed());
+        }
+    }
+    filter.expect("No server functions found")
 }
 
-fn transform_rx(message: Result<Message, warp::Error>) -> Result<String, LiveViewError> {
-    // destructure the message into the buffer we got from warp
-    let msg = message
-        .map_err(|_| LiveViewError::SendingFailed)?
-        .into_bytes();
+pub fn serve_dioxus_application<P: Clone + Send + Sync + 'static>(
+    cfg: ServeConfig<P>,
+) -> BoxedFilter<(impl Reply,)> {
+    // Serve the dist folder and the index.html file
+    let serve_dir = warp::fs::dir("./dist");
 
-    // transform it back into a string, saving us the allocation
-    let msg = String::from_utf8(msg).map_err(|_| LiveViewError::SendingFailed)?;
-
-    Ok(msg)
+    register_server_fns(cfg.server_fn_route.unwrap_or_default())
+        .or(warp::path::end()
+            .and(warp::get())
+            .map(move || warp::reply::html(dioxus_ssr_html(&cfg))))
+        .or(serve_dir)
+        .boxed()
 }
 
-async fn transform_tx(message: String) -> Result<Message, warp::Error> {
-    Ok(Message::text(message))
+#[derive(Debug)]
+struct FailedToReadBody(String);
+
+impl warp::reject::Reject for FailedToReadBody {}
+
+#[derive(Debug)]
+struct RecieveFailed(String);
+
+impl warp::reject::Reject for RecieveFailed {}
+
+async fn server_fn_handler(
+    server_context: DioxusServerContext,
+    function: Arc<ServerFnTraitObj>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Result<Box<dyn warp::Reply>, warp::Rejection> {
+    // Because the future returned by `server_fn_handler` is `Send`, and the future returned by this function must be send, we need to spawn a new runtime
+    let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
+    spawn_blocking({
+        move || {
+            tokio::runtime::Runtime::new()
+                .expect("couldn't spawn runtime")
+                .block_on(async {
+                    let resp = match function(server_context, &body).await {
+                        Ok(serialized) => {
+                            // if this is Accept: application/json then send a serialized JSON response
+                            let accept_header =
+                                headers.get("Accept").and_then(|value| value.to_str().ok());
+                            let mut res = Response::builder();
+                            if accept_header == Some("application/json")
+                                || accept_header
+                                    == Some(
+                                        "application/\
+                                            x-www-form-urlencoded",
+                                    )
+                                || accept_header == Some("application/cbor")
+                            {
+                                res = res.status(StatusCode::OK);
+                            }
+
+                            let resp = match serialized {
+                                Payload::Binary(data) => res
+                                    .header("Content-Type", "application/cbor")
+                                    .body(Bytes::from(data)),
+                                Payload::Url(data) => res
+                                    .header(
+                                        "Content-Type",
+                                        "application/\
+                                        x-www-form-urlencoded",
+                                    )
+                                    .body(Bytes::from(data)),
+                                Payload::Json(data) => res
+                                    .header("Content-Type", "application/json")
+                                    .body(Bytes::from(data)),
+                            };
+
+                            Box::new(resp.unwrap())
+                        }
+                        Err(e) => report_err(e),
+                    };
+
+                    if resp_tx.send(resp).is_err() {
+                        eprintln!("Error sending response");
+                    }
+                })
+        }
+    });
+    resp_rx.await.map_err(|err| {
+        warp::reject::custom(RecieveFailed(format!("Failed to recieve response {err}")))
+    })
+}
+
+fn report_err<E: Error>(e: E) -> Box<dyn warp::Reply> {
+    Box::new(
+        Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(format!("Error: {}", e))
+            .unwrap(),
+    ) as Box<dyn warp::Reply>
 }

packages/server/src/lib.rs

@@ -11,6 +11,8 @@ pub mod prelude {
     pub use crate::adapters::axum_adapter::*;
     #[cfg(feature = "salvo")]
     pub use crate::adapters::salvo_adapter::*;
+    #[cfg(feature = "warp")]
+    pub use crate::adapters::warp_adapter::*;
     #[cfg(feature = "ssr")]
     pub use crate::serve::ServeConfig;
     pub use crate::server_fn::{DioxusServerContext, ServerFn};