tina-core 0.0.2

Tina platform
Documentation
//! Request转换

use crate::tina::data::app_error::AppError;
use crate::tina::data::json::ToJson;
use crate::tina::data::multipart::unsend::de::MultipartDeserializer;
use crate::tina::data::multipart::unsend::Multipart;
use crate::tina::data::AppResult;
use crate::tina::i18n::message::system_message::SystemMessage;
use crate::tina::server::application::Application;
use crate::tina::server::http::multipart::MultipartData;
use crate::tina::server::http::path::de::PathDeserializer;
use crate::tina::server::http::path::path::{Path, PathItem};
use crate::tina::server::http::request::{ReqJson, ReqMultipart, ReqParam, ReqPath};
use crate::tina::server::http::request::{ReqMetadata, RequestExt};
use crate::tina::server::session::Session;
use crate::tina::validator::Validated;
use crate::{app_error_from, i18n_string};
use bytes::Bytes;
use futures_util::future::LocalBoxFuture;
use futures_util::{FutureExt, StreamExt, TryStreamExt};
use http::header::CONTENT_LENGTH;
use http::HeaderMap;
use ntex::http::encoding::Decoder;
use ntex::http::{HttpMessage, Payload};
use ntex::web::{FromRequest, HttpRequest};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;

fn url_decode(param_str: &str) -> AppResult<String> {
    match urlencoding::decode(param_str) {
        Ok(s) => Ok(s.into_owned()),
        Err(e) => {
            return Err(crate::app_param_check_error_with_msg!(i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT), "{:?}", e));
        }
    }
}

impl FromRequest<AppError> for Application {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, _payload: &mut Payload) -> Self::Future {
        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        async move { Ok(application) }.boxed_local()
    }
}

impl FromRequest<AppError> for Session {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, _payload: &mut Payload) -> Self::Future {
        let req2 = req.clone();
        async move {
            let session = Session::from_http_request(&req2).await?;
            Ok(session)
        }
        .boxed_local()
    }
}

impl FromRequest<AppError> for ReqMetadata {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, _payload: &mut Payload) -> Self::Future {
        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        let req2 = req.clone();
        let req3 = req.clone();
        let application2 = application.to_owned();
        async move {
            let session = Session::from_http_request(&req2).await?;
            Ok(ReqMetadata::new(req3, application2, session))
        }
        .boxed_local()
    }
}

impl<D: DeserializeOwned + Serialize + Validated> FromRequest<AppError> for ReqJson<D> {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
        let json = if let Ok(Some(mime)) = req.mime_type() {
            mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)
        } else {
            false
        };
        if !json {
            return async { Err(crate::app_param_check_error_with_msg!(i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT),)) }
                .boxed_local();
        }

        let len = req.headers().get(&CONTENT_LENGTH).and_then(|l| l.to_str().ok()).and_then(|s| s.parse::<usize>().ok());

        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        let limit = application.max_payload_size;
        if let Some(content_length) = len {
            if content_length > limit {
                return async { Err(crate::app_param_check_error_with_msg!(i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_TOO_LONG),)) }
                    .boxed_local();
            }
        }

        let mut payload = Decoder::from_headers(payload.take(), req.headers());
        let mut body = Vec::with_capacity(8192);
        let application = application.to_owned();
        let req2 = req.clone();
        let req3 = req.clone();

        async move {
            while let Some(item) = payload.next().await {
                let chunk = item.map_err(|err| crate::app_system_error!("{:?}", err))?;
                if (body.len() + chunk.len()) > limit {
                    return Err(crate::app_param_check_error_with_msg!(i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_TOO_LONG),));
                } else {
                    body.extend_from_slice(&chunk);
                }
            }
            let route_config = req3.get_route_config()?;
            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_raw_param_lossy(String::from_utf8_lossy(body.as_slice()));
            }
            let session = Session::from_http_request(&req2).await?;
            let data = serde_json::from_slice::<D>(&body).map_err(|err| {
                crate::app_param_check_error_with_msg!(
                    i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_PARAM_ANALYZE_FAILED, ("reason", err.to_string().as_str())),
                    "{:?}. source: {}",
                    err,
                    String::from_utf8_lossy(&body),
                )
            })?;
            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_param_val(&data);
            }
            data.validate(&session, &route_config.validate_group).await?;
            Ok(ReqJson::new(data, application, session))
        }
        .boxed_local()
    }
}

impl<D: DeserializeOwned + Serialize + Validated> FromRequest<AppError> for ReqParam<D> {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
        let len = req.headers().get(&CONTENT_LENGTH).and_then(|l| l.to_str().ok()).and_then(|s| s.parse::<usize>().ok());

        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        let limit = application.max_payload_size;
        if let Some(content_length) = len {
            if content_length > limit {
                return async move {
                    Err(crate::app_param_check_error_with_msg!(
                        i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT),
                        "content is too long: {}",
                        content_length
                    ))
                }
                .boxed_local();
            }
        }

        let mut payload = Decoder::from_headers(payload.take(), req.headers());
        let mut body = Vec::with_capacity(8192);

        let req = req.clone();
        let application = application.to_owned();
        let req2 = req.clone();
        let req3 = req.clone();
        async move {
            let route_config = req2.get_route_config()?;
            while let Some(item) = payload.next().await {
                let chunk = item.map_err(|err| crate::app_system_error!("{:?}", err))?;
                if (body.len() + chunk.len()) > limit {
                    return Err(crate::app_param_check_error_with_msg!(
                        i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT),
                        "content overflow",
                    ));
                } else {
                    body.extend_from_slice(&chunk);
                }
            }

            let mut param_str = req.query_string().to_owned();
            let body_str = String::from_utf8(body).map_err(|err| crate::app_system_error!("{:?}", err))?;

            if !body_str.is_empty() {
                if !param_str.is_empty() {
                    param_str.push('&');
                }
                param_str.push_str(body_str.as_str());
            }
            let parse = url_decode(param_str.as_str())?;
            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_raw_param_lossy(Cow::Borrowed(parse.as_str()));
            }

            let session = Session::from_http_request(&req2).await?;

            let data: D = serde_qs::from_str::<D>(parse.as_str()).map(|val| Ok(val)).unwrap_or_else(move |e| {
                tracing::error!(
                    "Failed during Query extractor deserialization. \
                     Request path: {:?}, error: {:?}",
                    req.path(),
                    e
                );
                Err(crate::app_param_check_error_with_msg!(
                    i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_PARAM_ANALYZE_FAILED, ("reason", e.to_string().as_str())),
                    "{:?}",
                    e
                ))
            })?;

            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_param_val(&data);
            }
            data.validate(&session, &route_config.validate_group).await?;
            Ok(ReqParam::new(data, application, session))
        }
        .boxed_local()
    }
}

impl<D: DeserializeOwned + Serialize + Validated> FromRequest<AppError> for ReqPath<D> {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, _payload: &mut Payload) -> Self::Future {
        let req = req.clone();
        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        let req2 = req.clone();
        let req3 = req.clone();
        async move {
            let route_config = req3.get_route_config()?;
            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_raw_param(req.match_info().get_ref().to_string());
            }
            let session = Session::from_http_request(&req2).await?;
            let req_path = req.path();
            let match_info = req.match_info();
            let mut path = Path::new(match_info.get_ref());
            for (name, value) in match_info.iter() {
                path.add(name, PathItem::Borrowed(Cow::Borrowed(value)));
            }

            let data: D = Deserialize::deserialize(PathDeserializer::new(&path)).map_err(move |e| {
                tracing::debug!(
                    "Failed during Path extractor deserialization. \
                         Request path: {:?}",
                    req_path
                );
                app_error_from!(e)
            })?;
            if let Ok(Some(log_param)) = req3.get_request_log_param() {
                log_param.set_param_val(&data);
            }
            data.validate(&session, &route_config.validate_group).await?;
            let names = path.segments.into_iter().map(|v| v.0.into_owned()).collect::<Vec<String>>();
            Ok(ReqPath::new(names, data, application, session))
        }
        .boxed_local()
    }
}

impl<D: DeserializeOwned + Serialize + Validated> FromRequest<AppError> for ReqMultipart<D> {
    type Error = AppError;
    type Future = LocalBoxFuture<'static, Result<Self, AppError>>;

    #[inline]
    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
        let content_type = match req.mime_type() {
            Ok(v) => v,
            Err(err) => {
                tracing::error!("{}", err);
                None
            }
        };
        if content_type.is_none() {
            return async {
                Err(crate::app_param_check_error_with_msg!(i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT), "empty content-type",))
            }
            .boxed_local();
        }
        if let Some(mime) = content_type {
            if mime.subtype() != mime::FORM_DATA {
                return async move {
                    Err(crate::app_param_check_error_with_msg!(
                        i18n_string!(SystemMessage::ERROR_HTTP_REQUEST_FORMAT),
                        "invalid content-type: {}",
                        mime.to_string().as_str()
                    ))
                }
                .boxed_local();
            }
        }
        let application = match req.get_application() {
            Ok(app) => app,
            Err(err) => {
                return async { Err(err) }.boxed_local();
            }
        };
        let req2 = req.clone();
        let req3 = req.clone();
        let req4 = req.clone();
        let application3 = application.to_owned();
        let application4 = application.to_owned();
        let payload1 = Payload::take(payload).map_ok(|v| Bytes::from_iter(v.into_iter())).map_err(app_error_from!());
        async move {
            let route_config = req2.get_route_config()?;
            let session = Session::from_http_request(&req2).await?;
            let mut headers = HeaderMap::new();
            for (k, v) in req3.headers() {
                headers.insert(k.clone(), v.clone().into());
            }
            let payload = Multipart::new(&headers, req3.query_string(), payload1, application3);
            match MultipartData::from_multipart(payload).await {
                Ok(data) => {
                    if let Ok(Some(log_param)) = req4.get_request_log_param() {
                        log_param.set_raw_param(data.to_json_string());
                    }
                    let deserializer = MultipartDeserializer::new(data).map_err(app_error_from!())?;
                    match D::deserialize(deserializer) {
                        Ok(data) => {
                            if let Ok(Some(log_param)) = req4.get_request_log_param() {
                                log_param.set_param_val(&data);
                            }
                            data.validate(&session, &route_config.validate_group).await?;
                            Ok(ReqMultipart::new(data, application4, session))
                        }
                        Err(err) => Err(app_error_from!(err)),
                    }
                }
                Err(err) => Err(err),
            }
        }
        .boxed_local()
    }
}