tina-core 0.0.2

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

use crate::app_error_from;
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::actix_web::middleware::demo_handler::DemoHandler;
use crate::tina::server::http::actix_web::middleware::operation_log_handler::OperationLogHandler;
use crate::tina::server::http::actix_web::middleware::panic_handler::PanicHandler;
use crate::tina::server::http::actix_web::middleware::permission_handler::PermissionHandler;
use crate::tina::server::http::actix_web::middleware::request_init_handler::RequestInitHandler;
use crate::tina::server::http::actix_web::middleware::request_log_param_handler::RequestLogParamHandler;
use crate::tina::server::http::actix_web::middleware::token_handler::TokenHandler;
use crate::tina::server::http::actix_web::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 actix_server::ServerHandle;
use actix_web::guard::{Guard, GuardContext};
use actix_web::{App, HttpServer, Resource};
use futures_util::future::BoxFuture;
use futures_util::FutureExt;
use http::Method;
use indexmap::IndexMap;
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) {
        actix_web::rt::System::new().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::new();

        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.handle();
    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;
}

struct MethodGuard(Method);

impl Guard for MethodGuard {
    fn check(&self, ctx: &GuardContext<'_>) -> bool {
        ctx.head().method == self.0
    }
}

fn to_guard(method: &HttpMethod) -> MethodGuard {
    MethodGuard(method.to_method())
}

async fn process_system_signals(
    server: ServerHandle,
    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(),
    };
    #[cfg(windows)]
    {
        let fut1 = async move {
            if let Ok(_) = actix_web::rt::signal::ctrl_c().await {
                tracing::info!("处理系统{}信号...", "Ctrl+C中断");
                application.terminate();
                shutdown_hook.await;
                server.stop(true).await;
                actix_web::rt::time::sleep(Duration::from_millis(1000)).await;
                exit(0)
            }
        };
        fut1.await;
    }
    #[cfg(unix)]
    {
        let mut sig_int =
            actix_web::rt::signal::unix::signal(actix_web::rt::signal::unix::SignalKind::interrupt()).expect("new sig_int failed");
        let mut sig_term =
            actix_web::rt::signal::unix::signal(actix_web::rt::signal::unix::SignalKind::terminate()).expect("new sig_term failed");
        let mut sig_quit =
            actix_web::rt::signal::unix::signal(actix_web::rt::signal::unix::SignalKind::quit()).expect("new sig_quit failed");
        let fut1 = async move {
            if let Some(_) = sig_int.recv().await {
                tracing::info!("处理系统{}信号...", "Ctrl+C中断");
            }
        };
        let fut2 = async move {
            if let Some(_) = sig_term.recv().await {
                tracing::info!("处理系统{}信号...", "Kill中断");
            }
        };
        let fut3 = async move {
            if let Some(_) = sig_quit.recv().await {
                tracing::info!("处理系统{}信号...", "Quit中断");
            }
        };
        futures_util::pin_mut!(fut1);
        futures_util::pin_mut!(fut2);
        let fut4 = crate::futures::future::select(fut1, fut2);
        futures_util::pin_mut!(fut3);
        futures_util::pin_mut!(fut4);
        let fut5 = crate::futures::future::select(fut4, fut3);
        let fut6 = async move {
            application.terminate();
            shutdown_hook.await;
            server.stop(true).await;
            actix_web::rt::time::sleep(Duration::from_millis(1000)).await;
            exit(0)
        };
        futures_util::pin_mut!(fut6);
        let fut7 = fut5.then(|_| fut6);
        fut7.await;
    }
}