tina-core 0.0.2

//! 基于nacos的注册中心客户端
#![cfg(feature = "client-nacos")]

use std::{any::TypeId, fmt::Debug, sync::Arc};

use config::{builder::DefaultState, Config, ConfigBuilder};
use nacos_sdk::api::{
    config::{ConfigResponse, ConfigService, ConfigServiceBuilder},
    naming::{NamingService, NamingServiceBuilder},
    props::ClientProps,
};
use serde::de::DeserializeOwned;
use tokio::runtime::Runtime;
use tracing::Instrument;

use crate::{
    app_error_from, app_system_error,
    tina::{client::GrpcClientProps, data::AppResult, server::application::Application, util::string::AsStr},
};

use super::{GrpcClientPropsKey, IRegistryClient, RegistryClient, RegistryClientExtentionValue};

type RegistryConfigService = Box<dyn ConfigService + Send + Sync + 'static>;
type RegistryNamingService = Box<dyn NamingService + Send + Sync + 'static>;

/// 注册到注册中心的实例
pub type RegistryServiceInstance = nacos_sdk::api::naming::ServiceInstance;

#[derive(Debug)]
pub(crate) struct RegistryConfigChangeListener(Application, Arc<Runtime>);
impl nacos_sdk::api::config::ConfigChangeListener for RegistryConfigChangeListener {
    fn notify(&self, config_resp: ConfigResponse) {
        let namespace = config_resp.namespace();
        let data_id = config_resp.data_id();
        let group = config_resp.group();
        let content = config_resp.content().to_owned();
        let content_type = config_resp.content_type().to_owned();
        let md5 = config_resp.md5();
        let span = debug_span!("listen_nacos_config_change", namespace, data_id, group, content, content_type, md5);
        let rt = self.1.clone();
        let application = self.0.clone();
        let func = application.registry_config_change_fn.clone();
        rt.spawn(
            async move {
                let config = match content_type.to_lowercase().as_str().trim() {
                    "json" => {
                        tracing::info!("found registry json config change, system will process it");
                        Some(
                            ConfigBuilder::<DefaultState>::default()
                                .add_source(config::File::from_str(content.as_str(), config::FileFormat::Json))
                                .build()
                                .map_err(app_error_from!())?,
                        )
                    }
                    "yaml" | "yml" => {
                        tracing::info!("found registry yaml config change, system will process it");
                        Some(
                            ConfigBuilder::<DefaultState>::default()
                                .add_source(config::File::from_str(content.as_str(), config::FileFormat::Yaml))
                                .build()
                                .map_err(app_error_from!())?,
                        )
                    }
                    _ => {
                        tracing::info!(
                            "found registry config change, but type '{content_type}' is not json and yaml, system will not process it."
                        );
                        None
                    }
                };
                if let Some(config) = config {
                    application.update_default_config(&config).await?;
                    if let Some(func) = func.as_ref() {
                        func(config).await?;
                    }
                }
                Ok(()) as AppResult<()>
            }
            .instrument(span),
        );
    }
}

#[async_trait]
impl IRegistryClient for RegistryClient {
    /// 初始化
    async fn init(&mut self, application: &Application) -> AppResult<()> {
        let mut props =
            ClientProps::new().server_addr(self.server_addr.as_str()).app_name(self.app_name.as_str()).namespace(self.namespace.as_str());
        let mut enable_auth = false;
        if let Some(v) = self.username.as_ref() {
            props = props.auth_username(v);
            enable_auth = true;
        }
        if let Some(v) = self.password.as_ref() {
            props = props.auth_password(v);
            enable_auth = true;
        }

        let mut config_service_builder = ConfigServiceBuilder::new(props.clone());
        let mut naming_service_builder = NamingServiceBuilder::new(props);
        if enable_auth {
            config_service_builder = config_service_builder.enable_auth_plugin_http();
            naming_service_builder = naming_service_builder.enable_auth_plugin_http();
        }
        let config_service = config_service_builder.build().map_err(app_error_from!())?;
        let naming_service = naming_service_builder.build().map_err(app_error_from!())?;

        let config_service = Arc::new(Box::new(config_service) as RegistryConfigService);
        let naming_service = Arc::new(Box::new(naming_service) as RegistryNamingService);

        self.extensions.insert(TypeId::of::<ConfigServiceBuilder>(), config_service.clone() as RegistryClientExtentionValue);
        self.extensions.insert(TypeId::of::<NamingServiceBuilder>(), naming_service.clone() as RegistryClientExtentionValue);

        let registry_config = application.get_registry_config()?;
        if let Some(configs) = registry_config.config.as_ref() {
            self.config_infos = configs.clone();
        }

        let should_listen_config_change = !self.config_infos.is_empty();

        if should_listen_config_change {
            let rt = tokio::runtime::Builder::new_multi_thread()
                .worker_threads(1)
                .thread_name("Nacos配置监听线程")
                .enable_all()
                .build()
                .map_err(app_error_from!())?;
            let listener = Arc::new(RegistryConfigChangeListener(application.clone(), Arc::new(rt)));
            for info in self.config_infos.iter() {
                let data_id = info.data_id.to_owned();
                let group = info.group.to_owned();
                config_service.add_listener(data_id.clone(), group.clone(), listener.clone()).await.map_err(app_error_from!())?;
                tracing::info!("监听注册中心配置: data_id = {data_id}, group = {group}")
            }
        }
        Ok(())
    }
    /// 读取配置
    async fn get_config<C>(&self, _application: &Application, data_id: String, group: String) -> AppResult<Option<C>>
    where
        C: DeserializeOwned + Send + Sync + 'static,
    {
        let config = self.get_raw_config(_application, data_id, group).await?;
        match config {
            Some(config) => config.try_deserialize::<C>().map_err(app_error_from!()).map(Some),
            None => Ok(None),
        }
    }

    /// 读取配置
    async fn get_raw_config(&self, _application: &Application, data_id: String, group: String) -> AppResult<Option<Config>> {
        let service =
            self.extensions.get(&TypeId::of::<ConfigServiceBuilder>()).ok_or_else(|| app_system_error!("No ConfigService found"))?;
        let config_service =
            service.downcast_ref::<RegistryConfigService>().ok_or_else(|| app_system_error!("ConfigService Type Error"))?;
        let config = config_service.get_config(data_id, group).await.map_err(app_error_from!())?;
        match config.content_type().to_lowercase().as_str().trim() {
            "json" => {
                let content = config.content().trim();
                match content.is_empty() {
                    true => Ok(None),
                    false => Ok(Some(
                        ConfigBuilder::<DefaultState>::default()
                            .add_source(config::File::from_str(content, config::FileFormat::Json))
                            .build()
                            .map_err(app_error_from!())?,
                    )),
                }
            }
            "yaml" => {
                let content = config.content().trim();
                match content.is_empty() {
                    true => Ok(None),
                    false => Ok(Some(
                        ConfigBuilder::<DefaultState>::default()
                            .add_source(config::File::from_str(content, config::FileFormat::Yaml))
                            .build()
                            .map_err(app_error_from!())?,
                    )),
                }
            }
            s => Err(app_system_error!(
                "Invalid content-type of config: {s}, expect 'json'. data_id: {}, group: {}",
                config.data_id(),
                config.group()
            )),
        }
    }

    /// 注册服务
    async fn regist_grpc_service<S>(&self, application: &Application, service: S) -> AppResult<()>
    where
        S: Into<RegistryServiceInstance> + Send + Sync + 'static,
    {
        let naming_service =
            self.extensions.get(&TypeId::of::<NamingServiceBuilder>()).ok_or_else(|| app_system_error!("No NamingService found"))?;
        let naming_service =
            naming_service.downcast_ref::<RegistryNamingService>().ok_or_else(|| app_system_error!("NamingService Type Error"))?;
        let registry_config = application.get_registry_config()?;
        let name = registry_config.app_name.clone();
        let group = registry_config.group_name.clone();
        let instance = service.into();
        naming_service.register_instance(name, group, instance).await.map_err(app_error_from!())?;
        Ok(())
    }

    /// 反注册服务
    async fn deregist_grpc_service<S>(&self, application: &Application, service: S) -> AppResult<()>
    where
        S: Into<RegistryServiceInstance> + Send + Sync + 'static,
    {
        let naming_service =
            self.extensions.get(&TypeId::of::<NamingServiceBuilder>()).ok_or_else(|| app_system_error!("No NamingService found"))?;
        let naming_service =
            naming_service.downcast_ref::<RegistryNamingService>().ok_or_else(|| app_system_error!("NamingService Type Error"))?;
        let registry_config = application.get_registry_config()?;
        let name = registry_config.app_name.clone();
        let group = registry_config.group_name.clone();
        let instance = service.into();
        naming_service.deregister_instance(name, group, instance).await.map_err(app_error_from!())?;
        Ok(())
    }
    /// 批量注册服务
    async fn regist_batch_grpc_service<S>(&self, application: &Application, service: Vec<S>) -> AppResult<()>
    where
        S: Into<RegistryServiceInstance> + Send + Sync + 'static,
    {
        let naming_service =
            self.extensions.get(&TypeId::of::<NamingServiceBuilder>()).ok_or_else(|| app_system_error!("No NamingService found"))?;
        let naming_service =
            naming_service.downcast_ref::<RegistryNamingService>().ok_or_else(|| app_system_error!("NamingService Type Error"))?;
        let registry_config = application.get_registry_config()?;
        let name = registry_config.app_name.clone();
        let group = registry_config.group_name.clone();
        let instance = service.into_iter().map(|s| s.into()).collect();
        naming_service.batch_register_instance(name, group, instance).await.map_err(app_error_from!())?;
        Ok(())
    }
    /// 获取服务调用属性
    async fn get_grpc_client_props(
        &self,
        app_name: &str,
        service_name: &str,
        group: impl Into<String> + Send,
    ) -> AppResult<GrpcClientProps> {
        let key = GrpcClientPropsKey {
            app_name: app_name.to_string(),
            service_name: service_name.to_string(),
            group: group.into(),
        };
        match self.props.get(&key) {
            Some(v) => Ok(v.value().clone()),
            None => {
                let app_name = key.app_name.to_string();
                let service_name = key.service_name.to_string();
                let group_name = Some(key.group.to_string());
                let service = self
                    .extensions
                    .get(&TypeId::of::<NamingServiceBuilder>())
                    .ok_or_else(|| app_system_error!("No NamingService found"))?;
                let naming_service =
                    service.downcast_ref::<RegistryNamingService>().ok_or_else(|| app_system_error!("NamingService Type Error"))?;
                let service_instance = naming_service
                    .select_one_healthy_instance(app_name.to_string(), group_name, vec![], false)
                    .await
                    .map_err(app_error_from!())?;
                let props = GrpcClientProps {
                    address: format!("{}:{}", service_instance.ip(), service_instance.port()),
                    app_name,
                    service_name,
                };
                self.props.insert(key.clone(), props);
                self.props.get(&key).ok_or_else(|| app_system_error!("No GrpcClientProps found")).map(|v| v.clone())
            }
        }
    }
}

#[allow(unused)]
#[cfg(test)]
mod test {
    use nacos_sdk::api::{
        config::{ConfigService, ConfigServiceBuilder},
        naming::{NamingService, NamingServiceBuilder},
        props::ClientProps,
    };
    use tokio::runtime::Runtime;

    use crate::{
        app_error_from,
        tina::{
            data::{app_error::AppError, AppResult},
            log::LogConfig,
        },
    };

    fn new_current_thread_rt() -> Result<Runtime, Box<dyn std::error::Error>> {
        Ok(tokio::runtime::Builder::new_current_thread().enable_all().build()?)
    }

    fn new_multi_thread_rt() -> Result<Runtime, Box<dyn std::error::Error>> {
        Ok(tokio::runtime::Builder::new_multi_thread().enable_all().build()?)
    }

    #[test]
    #[ignore]
    fn test() -> Result<(), Box<dyn std::error::Error>> {
        let (sender, receiver) = std::sync::mpsc::channel::<()>();
        new_multi_thread_rt()?.block_on(async move {
            LogConfig::default().init().await?;
            let props = ClientProps::new()
                .server_addr("127.0.0.1:8848")
                .namespace("")
                .app_name("tina-core")
                .auth_username("nacos")
                .auth_password("nacos");
            let config_service = ConfigServiceBuilder::new(props.clone()).enable_auth_plugin_http().build().map_err(app_error_from!())?;
            let res = config_service.get_config("test".to_string(), "DEFAULT_GROUP".to_string()).await.map_err(app_error_from!())?;
            let content = res.content();
            println!("{content}");

            let naming_service = NamingServiceBuilder::new(props).enable_auth_plugin_http().build().map_err(app_error_from!())?;
            sender.send(()).map_err(app_error_from!())?;
            Ok(()) as Result<(), AppError>
        });

        receiver.recv()?;
        Ok(()) as Result<(), Box<dyn std::error::Error>>
    }

    #[test]
    #[ignore]
    fn test_reqwest_block_on() -> Result<(), Box<dyn std::error::Error>> {
        let rt2 = new_multi_thread_rt().expect("build rt failed");
        new_current_thread_rt()?.block_on(async {
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            // let res = reqwest::blocking::Client::new().get("https://www.baidu.com").send()?;
            // let content = res.text()?;
            // println!("{content}");
            let handle = tokio::runtime::Handle::current();
            match handle.runtime_flavor() {
                tokio::runtime::RuntimeFlavor::CurrentThread => {
                    let (sender, receiver) = std::sync::mpsc::channel::<()>();
                    rt2.spawn(async move {
                        let res = reqwest::Client::new().get("https://www.baidu.com").send().await.map_err(app_error_from!())?;
                        let content = res.text().await.map_err(app_error_from!())?;
                        println!("{content}");
                        sender.send(()).expect("send failed");
                        Ok(()) as AppResult<()>
                    });
                    receiver.recv().expect("receive failed");
                    Ok(())
                }
                tokio::runtime::RuntimeFlavor::MultiThread => futures::executor::block_on(async move {
                    let res = reqwest::Client::new().get("https://www.baidu.com").send().await?;
                    let content = res.text().await?;
                    println!("{content}");
                    Ok(())
                }),
                _ => unreachable!(),
            }
        })
    }
}