dce_hyper/
lib.rs

1use std::any::Any;
2use std::collections::HashMap;
3use std::convert::Infallible;
4use std::ops::{Deref, DerefMut};
5use std::sync::{Arc, LazyLock};
6
7use async_trait::async_trait;
8use dce_router::api::{Api, Handler, Methods as ApiMethod, MARK_PATH_PART_SEPARATOR};
9use dce_router::context::Context;
10use dce_router::protocol::{RoutableProtocol, RpMeta};
11use dce_router::router::{RouteMatch, Router};
12use dce_util::result::{DceError, DceResult, SERVICE_UNAVAILABLE};
13use http_body_util::combinators::BoxBody;
14use http_body_util::{BodyExt, Full};
15use hyper::body::{Body, Bytes, Incoming};
16use hyper::header::{HeaderValue, COOKIE};
17use hyper::{Method as HyperMethod, Request, Response, StatusCode};
18use tokio::sync::RwLock;
19
20const HEADER_CONTENT_TYPE: &str = "Content-Type";
21const HEADER_SID_KEY: &str = "X-Session-Id";
22const COOKIE_SID_KEY: &str = "Session-Id=";
23
24pub struct HyperProtocol {
25    meta: RpMeta<Request<Incoming>>,
26    resp: Response<BoxBody<Bytes, Infallible>>,
27}
28
29impl Deref for HyperProtocol {
30    type Target = RpMeta<Request<Incoming>>;
31
32    fn deref(&self) -> &Self::Target {
33        &self.meta
34    }
35}
36
37impl DerefMut for HyperProtocol {
38    fn deref_mut(&mut self) -> &mut Self::Target {
39        &mut self.meta
40    }
41}
42
43impl HyperProtocol {
44    pub fn sid(&self) -> Option<&str> {
45        let headers = self.raw().headers();
46        headers.get(HEADER_SID_KEY).and_then(|hv| hv.to_str().ok())
47            .or_else(|| headers.get(COOKIE).and_then(|cv| cv.to_str().ok())
48                .and_then(|cs| cs.split(';').find_map(|l| l.find(COOKIE_SID_KEY).map(|i| &l[i+COOKIE_SID_KEY.len()..]))))
49    }
50
51    fn new(req: Request<Incoming>, ctx_data: Option<HashMap<String, Box<dyn Any + Send>>>) -> Self {
52        Self { meta: RpMeta::new(req, ctx_data), resp: Response::new(BoxBody::default()) }
53    }
54
55    fn response(mut ctx: Context<HyperProtocol>, err: Option<DceError>) -> Result<Response<BoxBody<Bytes, Infallible>>, Infallible> {
56        if let Some(hv) = ctx.ctx_data_remove(HEADER_CONTENT_TYPE)
57            .and_then(|ct| ct.downcast_ref::<String>().and_then(|ct| HeaderValue::from_str(ct).ok())) {
58            ctx.resp.headers_mut().insert(HEADER_CONTENT_TYPE, hv);
59        }
60        if let Some(hv) = ctx.resp_sid().and_then(|sid| HeaderValue::from_str(sid).ok()) {
61            ctx.resp.headers_mut().insert(HEADER_SID_KEY, hv);
62        }
63        ctx.try_print_err(&err);
64        if let Some(err) = err {
65            // 只自动设置公开且合法的 HTTP 状态码
66            *ctx.resp.status_mut() = StatusCode::from_u16(if err.public && err.code > 0 && err.code < 600 { err.code } else { SERVICE_UNAVAILABLE } as u16)
67                .unwrap_or(StatusCode::SERVICE_UNAVAILABLE);
68        }
69        if ctx.resp.body().is_end_stream() {
70            *ctx.resp.body_mut() = Full::from(Bytes::from_owner(ctx.clear_buffer())).boxed();
71        }
72        let HyperProtocol{ resp, ..} = ctx.into_rp();
73        Ok(resp)
74    }
75
76    pub async fn handle(self, routed: DceResult<RouteMatch<'_, HyperProtocol>>) -> Result<Response<BoxBody<Bytes, Infallible>>, Infallible> {
77        let mut ctx = Context::new(self, routed.as_ref().ok().map(|r| r.api));
78        let err = match routed {
79            Ok(routed) => routed.api.handle(&mut ctx, routed).await.err(),
80            err => err.err(),
81        };
82        Self::response(ctx, err)
83    }
84}
85
86#[async_trait]
87impl RoutableProtocol for HyperProtocol {
88    type Req = Request<Incoming>;
89
90    fn path(&self) ->  &str {
91        self.raw().uri().path().trim_start_matches(MARK_PATH_PART_SEPARATOR)
92    }
93
94    async fn body(&mut self) -> DceResult<Bytes> {
95        self.raw_mut().collect().await.map(|c| c.to_bytes()).map_err(DceError::priv_err0)
96    }
97
98    fn match_api(&self, apis: &Vec<&'static Api<Self>>) -> Option<&'static Api<Self>> {
99        apis.iter().find(|a| {
100            let method = hyper_to_api_method(self.raw().method());
101            a.methods & method == method && {
102                let hosts = a.hosts();
103                hosts.is_empty() || hosts.iter().any(|h| self.raw().uri().host().is_some_and(|rh|
104                    if h.contains(':') {
105                        rh.eq_ignore_ascii_case(h)
106                    } else if let Ok(port) = h.parse::<usize>() {
107                        rh.ends_with(format!(":{}", port).as_str())
108                    } else {
109                        rh.starts_with(format!("{}:", h).as_str())
110                    }
111                ))
112            }
113        }).map(|a| *a)
114    }
115}
116
117#[allow(non_snake_case, non_upper_case_globals)]
118pub mod Method {
119    use dce_router::api::Methods;
120
121    pub const Get: Methods = Methods(1);
122    pub const Post: Methods = Methods(2);
123    pub const Put: Methods = Methods(4);
124    pub const Delete: Methods = Methods(8);
125    pub const Head: Methods = Methods(16);
126    pub const Options: Methods = Methods(32);
127    pub const Connect: Methods = Methods(64);
128    pub const Patch: Methods = Methods(128);
129    pub const Trace: Methods = Methods(256);
130}
131
132const MAPPING: &[(HyperMethod, ApiMethod)] = &[
133    (HyperMethod::GET, Method::Get),
134    (HyperMethod::POST, Method::Post),
135    (HyperMethod::PUT, Method::Put),
136    (HyperMethod::DELETE, Method::Delete),
137    (HyperMethod::HEAD, Method::Head),
138    (HyperMethod::OPTIONS, Method::Options),
139    (HyperMethod::CONNECT, Method::Connect),
140    (HyperMethod::PATCH, Method::Patch),
141    (HyperMethod::TRACE, Method::Trace),
142];
143
144fn hyper_to_api_method(value: &HyperMethod) -> ApiMethod {
145    MAPPING.iter().find(|(hm, _)| value.eq(hm))
146        .map_or(Method::Get, |&(_, m)| m)
147}
148
149pub struct HttpRouter<Rp: RoutableProtocol + 'static>(Router<Rp>);
150
151impl <Rp: RoutableProtocol> Deref for HttpRouter<Rp> {
152    type Target = Router<Rp>;
153
154    fn deref(&self) -> &Self::Target {
155        &self.0
156    }
157}
158
159impl <Rp: RoutableProtocol> DerefMut for HttpRouter<Rp> {
160    fn deref_mut(&mut self) -> &mut Self::Target {
161        &mut self.0
162    }
163}
164
165impl <Rp: RoutableProtocol> HttpRouter<Rp> {
166    pub fn new() -> Self {
167        Self(Router::new())
168    }
169
170    pub fn get(&mut self, path: &'static str, handler: Handler<Rp>) -> &mut Self {
171        self.push_method(Method::Get | Method::Head, path, handler)
172    }
173    
174    pub fn post(&mut self, path: &'static str, handler: Handler<Rp>) -> &mut Self {
175        self.push_method(Method::Post|Method::Options, path, handler)
176    }
177
178    pub fn put(&mut self, path: &'static str, handler: Handler<Rp>) -> &mut Self {
179        self.push_method(Method::Put|Method::Options, path, handler)
180    }
181
182    pub fn patch(&mut self, path: &'static str, handler: Handler<Rp>) -> &mut Self {
183        self.push_method(Method::Patch|Method::Options, path, handler)
184    }
185
186    pub fn delete(&mut self, path: &'static str, handler: Handler<Rp>) -> &mut Self {
187        self.push_method(Method::Delete|Method::Options, path, handler)
188    }
189
190    fn push_method(&mut self, method: ApiMethod, path: &'static str, handler: Handler<Rp>) -> &mut Self {
191        let api = Api::new(path).by_methods(method);
192        self.0.bind_api(api, handler);
193        self
194    }
195}
196
197#[allow(non_snake_case, non_upper_case_globals)]
198pub static HyperRouter: LazyLock<Arc<RwLock<HttpRouter<HyperProtocol>>>> = LazyLock::new(|| Arc::new(RwLock::new(HttpRouter::<HyperProtocol>::new())));
199
200pub async fn hyper_route(req: Request<Incoming>, ctx_data: Option<HashMap<String, Box<dyn Any + Send>>>) -> Result<Response<BoxBody<Bytes, Infallible>>, Infallible> {
201    let router = HyperRouter.clone();
202    let router = router.read().await;
203    let rp = HyperProtocol::new(req, ctx_data);
204    let routed = router.lookup(&rp);
205    rp.handle(routed).await
206}
dce_hyper/lib.rs

dce_hyper/
lib.rs
