Skip to main content

threadloom_server/
lib.rs

1#[cfg(feature = "lambda")]
2pub use vercel_runtime;
3#[cfg(feature = "lambda")]
4pub use tokio;
5
6pub use http;
7use http::{Request, Response};
8use std::sync::Arc;
9use std::future::Future;
10use std::pin::Pin;
11
12pub type Bytes = Vec<u8>;
13pub type PortableRequest = Request<Bytes>;
14pub type PortableResponse = Response<Bytes>;
15
16pub trait Handler: Send + Sync + 'static {
17    fn handle(&self, req: PortableRequest) -> Pin<Box<dyn Future<Output = PortableResponse> + Send + '_>>;
18}
19
20pub struct Server {
21    pub routes: Vec<(String, Arc<dyn Handler>)>,
22}
23
24impl Default for Server {
25    fn default() -> Self {
26        Self::new()
27    }
28}
29
30impl Server {
31    pub fn new() -> Self {
32        Self { routes: Vec::new() }
33    }
34
35    pub fn route<H: Handler>(&mut self, path: &str, handler: H) {
36        self.routes.push((path.to_string(), Arc::new(handler)));
37    }
38}
39
40#[cfg(feature = "actix")]
41impl Server {
42    pub async fn run(self, port: u16) -> std::io::Result<()> {
43        let server = Arc::new(self);
44        actix_web::HttpServer::new(move || {
45            actix_web::App::new()
46                .configure(|cfg| actix_adapter::configure_arc(&server, cfg))
47        })
48        .bind(("0.0.0.0", port))?
49        .run()
50        .await
51    }
52
53    pub async fn run_with_actix_config<F>(self, port: u16, actix_cfg: F) -> std::io::Result<()>
54    where
55        F: Fn(&mut actix_web::web::ServiceConfig) + Send + Clone + 'static,
56    {
57        let server = Arc::new(self);
58        actix_web::HttpServer::new(move || {
59            actix_web::App::new()
60                .configure(|cfg| actix_adapter::configure_arc(&server, cfg))
61                .configure(actix_cfg.clone())
62        })
63        .bind(("0.0.0.0", port))?
64        .run()
65        .await
66    }
67}
68
69#[cfg(feature = "actix")]
70pub mod actix_adapter {
71    use super::{Handler, Server};
72    use actix_web::{web, HttpRequest, HttpResponse};
73    use std::sync::Arc;
74
75
76    async fn actix_handler_wrapper(
77        req: HttpRequest,
78        body: web::Bytes,
79        handler: web::Data<Arc<dyn Handler>>,
80    ) -> HttpResponse {
81        let method = http::Method::try_from(req.method().as_str()).unwrap();
82        
83        let mut builder = http::Request::builder()
84            .method(method)
85            .uri(req.uri().to_string());
86
87        for (k, v) in req.headers() {
88            builder = builder.header(k.as_str(), v.as_bytes());
89        }
90
91        let portable_req = builder.body(body.to_vec()).unwrap();
92        let portable_res = handler.handle(portable_req).await;
93
94        let mut res_builder = HttpResponse::build(
95            actix_web::http::StatusCode::from_u16(portable_res.status().as_u16()).unwrap(),
96        );
97
98        for (k, v) in portable_res.headers() {
99            res_builder.append_header((k.as_str(), v.as_bytes()));
100        }
101
102        res_builder.body(portable_res.into_body())
103    }
104
105    pub fn configure(server: &Server, cfg: &mut actix_web::web::ServiceConfig) {
106        for (path, handler) in &server.routes {
107            let handler_data = web::Data::new(Arc::clone(handler));
108            cfg.service(
109                web::resource(path)
110                    .app_data(handler_data.clone())
111                    .route(web::post().to(actix_handler_wrapper))
112            );
113        }
114    }
115
116    pub fn configure_arc(server: &Arc<Server>, cfg: &mut actix_web::web::ServiceConfig) {
117        for (path, handler) in &server.routes {
118            let handler_data = web::Data::new(Arc::clone(handler));
119            cfg.service(
120                web::resource(path)
121                    .app_data(handler_data.clone())
122                    .route(web::post().to(actix_handler_wrapper))
123            );
124        }
125    }
126}
127
128#[cfg(feature = "lambda")]
129pub mod lambda_adapter {
130    use super::Server;
131    use vercel_runtime::{run as vercel_run, AppState, Error, ResponseBody};
132    use std::future::Future;
133    use std::pin::Pin;
134    use std::sync::Arc;
135    use std::task::{Context, Poll};
136
137    type HyperRequest = hyper::Request<hyper::body::Incoming>;
138    type HyperResponse = hyper::Response<ResponseBody>;
139
140    struct ThreadloomService {
141        server: Arc<Server>,
142    }
143
144    impl Clone for ThreadloomService {
145        fn clone(&self) -> Self {
146            Self { server: Arc::clone(&self.server) }
147        }
148    }
149
150    impl tower::Service<(AppState, HyperRequest)> for ThreadloomService {
151        type Response = HyperResponse;
152        type Error = Error;
153        type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
154
155        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
156            Poll::Ready(Ok(()))
157        }
158
159        fn call(&mut self, (_, req): (AppState, HyperRequest)) -> Self::Future {
160            let server = Arc::clone(&self.server);
161            Box::pin(async move {
162                use http_body_util::BodyExt;
163
164                let path = if let Some(route) = req.headers().get("x-threadloom-route") {
165                    route.to_str().unwrap_or(req.uri().path()).to_string()
166                } else {
167                    req.uri().path().to_string()
168                };
169
170                let (parts, body) = req.into_parts();
171                let bytes = body.collect().await
172                    .map(|c| c.to_bytes().to_vec())
173                    .unwrap_or_default();
174
175                let found_handler = server.routes.iter()
176                    .find(|(route_path, _)| route_path == &path)
177                    .map(|(_, h)| Arc::clone(h));
178
179                if let Some(h) = found_handler {
180                    let portable_req = http::Request::from_parts(parts, bytes);
181                    let res = h.handle(portable_req).await;
182                    let (res_parts, res_bytes) = res.into_parts();
183                    Ok(hyper::Response::from_parts(res_parts, ResponseBody::from(res_bytes)))
184                } else {
185                    Ok(hyper::Response::builder()
186                        .status(404)
187                        .body(ResponseBody::from("Not Found"))
188                        .unwrap())
189                }
190            })
191        }
192    }
193
194    pub async fn run(server: Server) -> Result<(), Error> {
195        let service = ThreadloomService { server: Arc::new(server) };
196        vercel_run(service).await
197    }
198}
199
200
201#[cfg(not(feature = "lambda"))]
202pub mod lambda_http {
203    #[derive(Debug)]
204    pub struct Error;
205}
206
207#[cfg(not(feature = "lambda"))]
208pub mod tokio {
209    pub mod runtime {
210        pub struct Builder;
211        impl Builder {
212            pub fn new_current_thread() -> Self { Self }
213            pub fn enable_all(&mut self) -> &mut Self { self }
214            pub fn build(&mut self) -> Result<Runtime, ()> { Ok(Runtime) }
215        }
216        pub struct Runtime;
217        impl Runtime {
218            pub fn block_on<F>(&self, _f: F) -> Result<(), crate::lambda_http::Error> { Ok(()) }
219        }
220    }
221}
222
223#[cfg(not(feature = "lambda"))]
224pub mod lambda_adapter {
225    use super::Server;
226    pub async fn run(_server: Server) -> Result<(), Box<dyn std::error::Error>> {
227        Ok(())
228    }
229}