#![cfg(feature = "server-tonic")]
#![allow(dead_code, missing_docs)]
use crate::tina::server::{random_address_and_port, Server};
use crate::tina::{
server::application::{AppConfig, Application},
util::not_empty::INotEmpty,
};
use config::Config;
use futures_util::FutureExt;
use std::net::SocketAddr;
use std::process::exit;
use std::str::FromStr;
use std::sync::mpsc::{channel, Sender};
use std::time::Duration;
use self::{
middleware::{request_init_handler::RequestInitHandler, routes_handler::RoutesHandler},
service_ext::{GrpcServiceRegister, NoService},
};
use super::{super::ServerShutdownHook, IGrpcServer};
pub mod middleware;
pub mod request_ext;
pub mod response_ext;
pub mod service_ext;
pub mod session_ext;
impl IGrpcServer for Server {
fn run_grpc(self, config: Config) {
let (application, config) =
AppConfig::from_config(config).unwrap_or_else(|err| panic!("parse config to application failed: {:?}", err));
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(application.workers)
.thread_name("tonic-worker-thread")
.enable_all()
.build()
.expect("create tokio runtime failed");
let (shutdown_sender, shutdown_receiver) = channel::<()>();
runtime.spawn(run_async(application, config, shutdown_sender, self));
shutdown_receiver.recv().expect("receive shutdown signal failed");
runtime.shutdown_background();
}
}
async fn run_async(mut application: AppConfig, config: Config, shutdown_sender: Sender<()>, server: Server) {
let Server {
init,
post_init,
after_start,
shutdown_hook,
} = server;
if let Some(init_fn) = init {
let config2 = config.clone();
let app_config = match init_fn(application, config2).await {
Ok(v) => v,
Err(err) => {
println!("init app config failed: {:?}", err);
tracing::error!("init app config failed: {:?}", err);
exit(-1)
}
};
application = app_config;
}
let application = match application.init(&config).await {
Ok(v) => v,
Err(err) => {
println!("application init failed: {:?}", err);
exit(-1);
}
};
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)
}
}
}
let bind_address = match application.address.not_empty() && application.port.is_some() {
true => {
let address = application.address.as_str();
let port = application.port.unwrap_or_default();
SocketAddr::from_str(format!("{}:{}", address, port).as_str())
.unwrap_or_else(|err| panic!("绑定端口失败: {}:{}, reason: {:?}", address, port, err))
}
false => random_address_and_port(application.address.as_str()),
};
let application_for_after_start: Application = application.to_owned();
let application_for_signal: Application = application.to_owned();
let router = GrpcServiceRegister::get_router();
let routes = router.into_service();
tonic::transport::Server::builder()
.layer(RequestInitHandler::new(&application))
.layer(RoutesHandler::new(routes))
.add_optional_service(None as Option<NoService>)
.serve_with_shutdown(bind_address, async move {
#[cfg(feature = "client-nacos")]
{
self::nacos::regist_service(&application, bind_address).await;
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
tracing::info!("启动Tonic Server成功 {}", bind_address);
match after_start {
Some(after_start) => {
match after_start(application_for_after_start).await {
Ok(_) => {
process_system_signals(application_for_signal, bind_address, shutdown_hook).await;
}
Err(err) => {
tracing::error!("Failed to process after server started: {err:?}");
}
}
shutdown_sender.send(()).expect("send shutdown signal failed");
}
None => {
process_system_signals(application_for_signal, bind_address, shutdown_hook).await;
shutdown_sender.send(()).expect("send shutdown signal failed");
}
}
})
.await
.expect("启动Tonic Server失败!");
}
async fn process_system_signals(application: Application, bind_address: SocketAddr, shutdown_hook: Option<ServerShutdownHook>) {
#[allow(unused_variables)]
let application2 = application.clone();
let hook_application = application.clone();
let shutdown_hook = match shutdown_hook {
None => async move {
#[cfg(feature = "client-nacos")]
{
nacos::deregist_service(&application2, bind_address).await;
}
}
.boxed(),
Some(hook) => async move {
#[cfg(feature = "client-nacos")]
{
nacos::deregist_service(&application2, bind_address).await;
}
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 application.shutdown_after_start.unwrap_or_default() {
shutdown_hook.await;
tokio::time::sleep(Duration::from_millis(1000)).await;
application.terminate();
return;
}
let ctrl_c = async {
tokio::signal::ctrl_c().await.expect("failed to install Ctrl+C handler");
};
#[cfg(unix)]
let terminate = async {
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("failed to install terminate signal handler")
.recv()
.await;
};
#[cfg(unix)]
let interrupt = async {
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
.expect("failed to install interrupt signal handler")
.recv()
.await;
};
#[cfg(unix)]
let quit = async {
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::quit()).expect("failed to install quit signal handler").recv().await;
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
#[cfg(not(unix))]
let interrupt = std::future::pending::<()>();
#[cfg(not(unix))]
let quit = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {
tracing::info!("处理系统{}信号...", "Ctrl+C中断");
shutdown_hook.await;
tokio::time::sleep(Duration::from_millis(1000)).await;
application.terminate();
},
_ = terminate => {
tracing::info!("处理系统{}信号...", "Kill中断");
shutdown_hook.await;
tokio::time::sleep(Duration::from_millis(1000)).await;
application.terminate();
},
_ = interrupt => {
tracing::info!("处理系统{}信号...", "Ctrl+C中断");
shutdown_hook.await;
tokio::time::sleep(Duration::from_millis(1000)).await;
application.terminate();
},
_ = quit => {
tracing::info!("处理系统{}信号...", "Ctrl+C中断");
shutdown_hook.await;
tokio::time::sleep(Duration::from_millis(1000)).await;
application.terminate();
},
}
}
#[cfg(feature = "client-nacos")]
mod nacos {
use std::{collections::HashMap, net::SocketAddr, process::exit};
use crate::tina::{
client::registry::{IRegistryClient, RegistryServiceInstance},
data::AppResult,
server::application::Application,
};
pub(crate) async fn regist_service(application: &Application, address: SocketAddr) {
let application = application.clone();
if let Err(err) = tokio::spawn(async move {
let func = move || async move {
let client = application.get_registry_client()?;
let registry_config = application.get_registry_config()?;
let healthy = registry_config.healthy.unwrap_or(true);
let instance = RegistryServiceInstance {
instance_id: None,
ip: address.ip().to_string(),
port: address.port() as i32,
weight: 1.0,
healthy,
enabled: true,
ephemeral: false,
cluster_name: None,
service_name: None,
metadata: HashMap::new(),
};
client.regist_grpc_service(&application, instance.clone()).await?;
tracing::info!("注册服务成功: {instance:?}");
Ok(()) as AppResult<()>
};
loop {
let func2 = func.clone();
if let Err(err) = func2().await {
tracing::error!("注册服务失败: {err:?}");
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
continue;
}
break;
}
})
.await
{
tracing::error!("注册服务失败: {err:?}");
exit(-1);
}
}
pub(crate) async fn deregist_service(application: &Application, address: SocketAddr) {
let application = application.clone();
if let Err(err) = tokio::spawn(async move {
let func = move || async move {
let client = application.get_registry_client()?;
let registry_config = application.get_registry_config()?;
let healthy = registry_config.healthy.unwrap_or(true);
let instance = RegistryServiceInstance {
instance_id: None,
ip: address.ip().to_string(),
port: address.port() as i32,
weight: 1.0,
healthy,
enabled: true,
ephemeral: false,
cluster_name: None,
service_name: None,
metadata: HashMap::new(),
};
client.deregist_grpc_service(&application, instance.clone()).await?;
tracing::info!("反注册服务成功: {instance:?}");
Ok(()) as AppResult<()>
};
loop {
let func2 = func.clone();
if let Err(err) = func2().await {
tracing::error!("反注册服务失败: {err:?}");
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
continue;
}
break;
}
})
.await
{
tracing::error!("注册服务失败: {err:?}");
exit(-1);
}
}
}
#[cfg(test)]
mod test {
#![allow(unused)]
use tokio::runtime::Runtime;
fn rt() -> Result<Runtime, Box<dyn std::error::Error>> {
Ok(tokio::runtime::Builder::new_multi_thread().enable_all().build()?)
}
#[test]
fn test() -> Result<(), Box<dyn std::error::Error>> {
rt()?.block_on(async move { Ok(()) })
}
}