fusen_rs/server/
router.rs

1use crate::{
2    error::FusenError,
3    handler::HandlerContext,
4    protocol::{
5        codec::{FusenHttpCodec, RequestCodec, ResponseCodec},
6        fusen::{context::FusenContext, metadata::MetaData, request::FusenRequest},
7    },
8    server::{
9        path::{PathCache, QueryResult},
10        rpc::RpcServerHandler,
11    },
12};
13use bytes::Bytes;
14use fusen_internal_common::{BoxFuture, utils::uuid::uuid};
15use http::{Request, Response};
16use http_body_util::{BodyExt, Full, combinators::BoxBody};
17use hyper::service::Service;
18use std::{convert::Infallible, sync::Arc};
19
20#[derive(Clone)]
21pub struct Router {
22    pub context: Arc<RouterContext>,
23}
24
25pub struct RouterContext {
26    pub http_codec: FusenHttpCodec,
27    pub path_cache: PathCache,
28    pub handler_context: HandlerContext,
29    pub fusen_service_handler: RpcServerHandler,
30}
31
32impl Service<Request<hyper::body::Incoming>> for Router {
33    type Response = Response<BoxBody<Bytes, Infallible>>;
34
35    type Error = FusenError;
36
37    type Future = BoxFuture<Result<Self::Response, Self::Error>>;
38
39    fn call(&self, request: Request<hyper::body::Incoming>) -> Self::Future {
40        let router = self.context.clone();
41        Box::pin(async move {
42            let result = call(request, router).await;
43            match result {
44                Ok(response) => Ok(response),
45                Err(error) => Ok(error.into()),
46            }
47        })
48    }
49}
50
51async fn call(
52    request: Request<hyper::body::Incoming>,
53    router: Arc<RouterContext>,
54) -> Result<Response<BoxBody<Bytes, Infallible>>, FusenError> {
55    //首先进行编解码
56    let request = request.map(|e| e.boxed());
57    let mut fusen_request: FusenRequest = RequestCodec::decode(&router.http_codec, request).await?;
58    //通过path找到资源
59    let Some(QueryResult {
60        method_info,
61        rest_fields,
62    }) = router.path_cache.seach(&fusen_request.path).await
63    else {
64        return Response::builder()
65            .status(404)
66            .body(Full::new(Bytes::new()).boxed())
67            .map_err(|error| FusenError::Error(Box::new(error)));
68    };
69    if let Some(rest_fields) = rest_fields {
70        for (key, value) in rest_fields {
71            fusen_request.querys.insert(key, value);
72        }
73    }
74    let context = FusenContext {
75        unique_identifier: uuid(),
76        metadata: MetaData::default(),
77        method_info,
78        request: fusen_request,
79        response: Default::default(),
80    };
81    //通过service获取handler
82    let handler_controller = router
83        .handler_context
84        .get_controller(&context.method_info.service_desc)?;
85    let aspect_handers = handler_controller.aspect.clone();
86    let context = router
87        .fusen_service_handler
88        .call(aspect_handers, context)
89        .await?;
90    let response = ResponseCodec::encode(
91        &router.http_codec,
92        &mut context
93            .response
94            .ok_or(FusenError::ErrorMessage("fusen_service_handler call error"))?,
95    )?;
96    Ok(response)
97}
98
99impl From<FusenError> for Response<BoxBody<Bytes, Infallible>> {
100    fn from(error: FusenError) -> Self {
101        let mut builder = Response::builder();
102        let mut body = Bytes::new();
103        match error {
104            FusenError::HttpError(http_status) => {
105                builder = builder.status(http_status.status);
106                if let Some(message) = http_status.message {
107                    body = Bytes::copy_from_slice(message.as_bytes());
108                }
109            }
110            _error => {
111                builder = builder.status(500);
112            }
113        }
114        builder
115            .body(Full::new(body).boxed())
116            //
117            .unwrap_or(Response::new(Full::new(Bytes::new()).boxed()))
118    }
119}
fusen_rs/server/router.rs

fusen_rs/server/
router.rs
