tina-core 0.0.2

//! 基于ntex的server mod
#![cfg(feature = "server-ntex")]
#![allow(dead_code)]

use crate::tina::api_doc;
use crate::tina::core::service::lock_service::IDistributedLockService;
use crate::tina::data::AppResult;
use crate::tina::redis::lock_service::RedisDistributedLock;
use crate::tina::redis::IRedisClient;
use crate::tina::server::application::{AppConfig, Application};
use crate::tina::server::http::ntex::middleware::demo_handler::DemoHandler;
use crate::tina::server::http::ntex::middleware::operation_log_handler::OperationLogHandler;
use crate::tina::server::http::ntex::middleware::panic_handler::PanicHandler;
use crate::tina::server::http::ntex::middleware::permission_handler::PermissionHandler;
use crate::tina::server::http::ntex::middleware::request_init_handler::RequestInitHandler;
use crate::tina::server::http::ntex::middleware::request_log_param_handler::RequestLogParamHandler;
use crate::tina::server::http::ntex::middleware::token_handler::TokenHandler;
use crate::tina::server::http::ntex::middleware::transaction_handler::TransactionHandler;
use crate::tina::server::http::route::RouteConfigKey;
use crate::tina::server::http::route::RouteRegister;
use crate::tina::server::http::route::{HttpMethod, RouteBaseConfig};
use crate::tina::server::{IHttpServer, Server};
use crate::{app_error_from, app_system_error};
use futures_util::future::BoxFuture;
use futures_util::FutureExt;
use indexmap::IndexMap;
use ntex::rt::Signal;
use ntex::server::Server;
use ntex::web::guard::MethodGuard;
use ntex::web::{App, HttpServer, Resource};
use std::ops::Deref;
use std::process::exit;
use std::sync::Arc;
use std::time::Duration;

pub mod app_error_ext;
pub mod application_ext;
pub(in crate::tina::server) mod delegate;
pub(in crate::tina::server) mod middleware;
pub mod multipart;
pub mod request_conversion;
pub mod request_ext;
pub mod response_data_ext;
pub mod response_ext;
pub mod response_page_ext;
pub mod response_stream_ext;
pub mod route_ext;
pub mod session_ext;

impl IHttpServer for Server {
    fn run_http(self, application: AppConfig) {
        ntex::rt::System::build().name("Server启动").stop_on_panic(false).finish().block_on(run_async(application, self));
    }
}

async fn run_async(mut application: AppConfig, server: Server) {
    let Server {
        workers,
        address,
        port,
        init,
        post_init,
        shutdown_hook,
    } = server;

    if let Err(err) = application.init().await {
        println!("application init failed: {:?}", err);
        exit(-1);
    }
    if let Some(init_fn) = init {
        let config = match async move {
            let redis_client = application.get_redis_client().map_err(app_error_from!())?;
            let mut lock = RedisDistributedLock(redis_client, None);
            lock.lock("application_start", Some("application_start"), Duration::from_secs(60), Duration::from_secs(5), async move {
                init_fn(application).await
            })
            .await
        }
        .await
        {
            Ok(v) => v,
            Err(err) => {
                println!("init app config failed: {:?}", err);
                tracing::error!("init app config failed: {:?}", err);
                exit(-1)
            }
        };
        application = config;
    }

    let application = Application::from(application);
    if let Some(post_init_fn) = post_init {
        let application2 = application.clone();
        match async move { post_init_fn(application2).await }.await {
            Ok(_) => {}
            Err(err) => {
                println!("post init app config failed: {:?}", err);
                tracing::error!("post init app config failed: {:?}", err);
                exit(-1)
            }
        }
    }

    if application.enable_api_doc {
        api_doc::init_api_doc(&application)
    }
    let application_for_signal: Application = application.to_owned();
    let application = Arc::new(application);

    let config_map: IndexMap<RouteConfigKey, Arc<RouteBaseConfig>> = RouteRegister::get_registed_routes()
        .into_iter()
        .map(|config| {
            let method = config.config.method.clone();
            let base_config = Arc::new(config.config);
            let path = RouteBaseConfig::get_route_path(application.deref(), &base_config);
            (
                RouteConfigKey {
                    method,
                    path,
                },
                base_config,
            )
        })
        .collect::<IndexMap<RouteConfigKey, Arc<RouteBaseConfig>>>();

    let server = HttpServer::new(move || {
        let application2 = application.deref().to_owned();
        let mut route_configs = RouteRegister::get_registed_routes();
        let mut app = App::with(app_system_error!("App Error"));

        route_configs.sort_by(|o1, o2| {
            let path1 = RouteBaseConfig::get_route_path(&application2, &o1.config);
            let path2 = RouteBaseConfig::get_route_path(&application2, &o2.config);
            path1.cmp(&path2)
        });

        for config in route_configs.into_iter() {
            let method = config.config.method.clone();
            let base = config.config;
            if let Some(route) = config.route {
                let path = RouteBaseConfig::get_route_path(application.deref(), &base);
                let base_config = config_map
                    .get(&RouteConfigKey {
                        method: method.clone(),
                        path: path.clone(),
                    })
                    .unwrap_or_else(|| panic!("get base config failed: {}", path));
                // 注意这里的中间件的执行顺序是: 执行接口函数之前的部分(从下往上执行), 执行接口函数之后的部分(从上往下执行)
                let resource = Resource::new(path.as_str())
                    .wrap(PermissionHandler) // 基于RBAC的访问控制
                    .wrap(TokenHandler) // 用户登录检查
                    .wrap(TransactionHandler) // 自动事务处理
                    .wrap(PanicHandler) // panic捕获处理
                    .wrap(RequestLogParamHandler) // 请求参数日志处理
                    .wrap(OperationLogHandler) // 操作日志处理
                    .wrap(DemoHandler) // 演示模式处理
                    .wrap(PanicHandler) // panic捕获处理
                    .wrap(RequestInitHandler::new(&application2, base_config)) // 请求信息初始化处理
                    .guard(to_guard(&method))
                    .route(route);
                app = app.service(resource);
            }
        }
        app
    })
    .workers(workers)
    .bind(format!("{}:{}", address, port))
    .map_err(|err| crate::app_system_error!("绑定端口失败: {}", err))
    .expect("create http server failed")
    .run();

    let server1 = server.clone();
    let fut1 = async move {
        server.await.expect("start http server falied");
        ()
    };
    crate::futures::future::join(process_system_signals(server1, application_for_signal, shutdown_hook), fut1).await;
}

fn to_guard(method: &HttpMethod) -> MethodGuard {
    match method {
        HttpMethod::OPTIONS => ntex::web::guard::Options(),
        HttpMethod::GET => ntex::web::guard::Get(),
        HttpMethod::POST => ntex::web::guard::Post(),
        HttpMethod::PUT => ntex::web::guard::Put(),
        HttpMethod::DELETE => ntex::web::guard::Delete(),
        HttpMethod::HEAD => ntex::web::guard::Head(),
        HttpMethod::TRACE => ntex::web::guard::Trace(),
        HttpMethod::CONNECT => ntex::web::guard::Connect(),
        HttpMethod::PATCH => ntex::web::guard::Patch(),
    }
}

async fn process_system_signals(
    server: Server,
    application: Application,
    shutdown_hook: Option<Box<dyn FnOnce(Application) -> BoxFuture<'static, AppResult<()>> + Send + 'static>>,
) {
    let hook_application = application.clone();
    let shutdown_hook = match shutdown_hook {
        None => async move { () }.boxed(),
        Some(hook) => async move {
            if let Err(err) = hook(hook_application).await {
                tracing::error!("{:?}", err);
            }
            if let Err(err) = crate::tina::log::wait_for_closeable_appender_shutdown() {
                println!("{:?}", err);
            }
            ()
        }
        .boxed(),
    };
    if let Some(receiver) = ntex::rt::signal() {
        if let Ok(s) = receiver.await {
            match s {
                Signal::Hup => {
                    tracing::info!("处理系统{}信号...", "Sign Hup中断");
                    application.terminate();
                    shutdown_hook.await;
                    server.stop(false).await; // 目前Linux系统下的优雅停机有BUG, 程序会卡在这一步, 所以暂时禁用Linux系统下的优雅停机
                    ntex::time::sleep(Duration::from_millis(1000)).await;
                    exit(0)
                }
                Signal::Int => {
                    tracing::info!("处理系统{}信号...", "Ctrl+C中断");
                    application.terminate();
                    shutdown_hook.await;
                    server.stop(true).await;
                    ntex::time::sleep(Duration::from_millis(1000)).await;
                    exit(0)
                }
                Signal::Term => {
                    tracing::info!("处理系统{}信号...", "Kill中断");
                    application.terminate();
                    shutdown_hook.await;
                    server.stop(false).await; // 目前Linux系统下的优雅停机有BUG, 程序会卡在这一步, 所以暂时禁用Linux系统下的优雅停机
                    ntex::time::sleep(Duration::from_millis(1000)).await;
                    exit(0)
                }
                Signal::Quit => {
                    tracing::info!("处理系统{}信号...", "Quit中断");
                    application.terminate();
                    shutdown_hook.await;
                    server.stop(false).await; // 目前Linux系统下的优雅停机有BUG, 程序会卡在这一步, 所以暂时禁用Linux系统下的优雅停机
                    ntex::time::sleep(Duration::from_millis(1000)).await;
                    exit(0)
                }
            }
        }
    }
}