refactor:优化锁使用粒度, 避免锁冲突导致资源获取超时

Author: chuntaojun

examples/discover.rs

@@ -40,7 +40,7 @@ async fn main() -> Result<(), PolarisError> {
         .with_level(true)
         .with_line_number(true)
         .with_thread_ids(true)
-        .with_max_level(LevelFilter::DEBUG)
+        .with_max_level(LevelFilter::INFO)
         // sets this to be the default, global collector for this application.
         .init();
 
@@ -116,13 +116,16 @@ async fn main() -> Result<(), PolarisError> {
         Ok(_) => {}
     }
 
-    let watch_rsp = consumer.watch_instance(WatchInstanceRequest {
-        namespace: "rust-demo".to_string(),
-        service: "polaris-rust-provider".to_string(),
-        call_back: Box::new(|instances| {
-            tracing::info!("watch instance: {:?}", instances.instances);
-        }),
-    }).await;
+    tracing::info!("begin do watch service_instances change");
+    let watch_rsp = consumer
+        .watch_instance(WatchInstanceRequest {
+            namespace: "rust-demo".to_string(),
+            service: "polaris-rust-provider".to_string(),
+            call_back: Arc::new(|instances| {
+                tracing::info!("watch instance: {:?}", instances.instances);
+            }),
+        })
+        .await;
 
     match watch_rsp {
         Err(err) => {
@@ -131,10 +134,6 @@ async fn main() -> Result<(), PolarisError> {
         Ok(_) => {}
     }
 
-    for _ in 0..120 {
-        std::thread::sleep(Duration::from_secs(1));
-    }
-
     let instances_ret = consumer
         .get_all_instance(GetAllInstanceRequest {
             flow_id: uuid::Uuid::new_v4().to_string(),
@@ -153,6 +152,10 @@ async fn main() -> Result<(), PolarisError> {
         }
     }
 
+    for _ in 0..120 {
+        std::thread::sleep(Duration::from_secs(1));
+    }
+
     // 反注册
     let deregister_req = InstanceDeregisterRequest {
         flow_id: uuid::Uuid::new_v4().to_string(),

src/config/api.rs

@@ -13,10 +13,77 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-pub trait ConfigFileAPI {
+use std::sync::Arc;
 
+use crate::{
+    config::default::DefaultConfigFileAPI,
+    core::{
+        context::SDKContext,
+        model::{config::ConfigFile, error::PolarisError},
+    },
+};
+
+use super::req::{
+    DeleteConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest, PublishConfigFileRequest,
+    UpdateConfigFileRequest, WatchConfigFileRequest, WatchConfigFileResponse,
+    WatchConfigGroupRequest, WatchConfigGroupResponse,
+};
+
+/// new_config_file_api
+pub fn new_config_file_api() -> Result<impl ConfigFileAPI, PolarisError> {
+    let start_time = std::time::Instant::now();
+    let context_ret = SDKContext::default();
+    tracing::info!("create sdk context cost: {:?}", start_time.elapsed());
+    if context_ret.is_err() {
+        return Err(context_ret.err().unwrap());
+    }
+    Ok(DefaultConfigFileAPI::new(
+        Arc::new(context_ret.unwrap()),
+        true,
+    ))
+}
+
+/// new_config_file_api_by_context
+pub fn new_config_file_api_by_context(
+    context: Arc<SDKContext>,
+) -> Result<impl ConfigFileAPI, PolarisError> {
+    Ok(DefaultConfigFileAPI::new(context, false))
 }
 
-pub trait ConfigGroupAPI {
+#[async_trait::async_trait]
+pub trait ConfigFileAPI
+where
+    Self: Send + Sync,
+{
+    async fn get_config_file(&self, req: GetConfigFileRequest) -> Result<ConfigFile, PolarisError>;
+
+    async fn update_config_file(&self, req: UpdateConfigFileRequest) -> Result<bool, PolarisError>;
+
+    async fn delete_config_file(&self, req: DeleteConfigFileRequest) -> Result<bool, PolarisError>;
 
-}
+    async fn publish_config_file(
+        &self,
+        req: PublishConfigFileRequest,
+    ) -> Result<bool, PolarisError>;
+
+    async fn watch_config_file(
+        &self,
+        req: WatchConfigFileRequest,
+    ) -> Result<WatchConfigFileResponse, PolarisError>;
+}
+
+#[async_trait::async_trait]
+pub trait ConfigGroupAPI
+where
+    Self: Send + Sync,
+{
+    async fn get_publish_config_files(
+        &self,
+        req: GetConfigGroupRequest,
+    ) -> Result<Vec<ConfigFile>, PolarisError>;
+
+    async fn watch_publish_config_files(
+        &self,
+        req: WatchConfigGroupRequest,
+    ) -> Result<WatchConfigGroupResponse, PolarisError>;
+}

src/config/default.rs

@@ -0,0 +1,104 @@
+// Tencent is pleased to support the open source community by making Polaris available.
+//
+// Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+//
+// Licensed under the BSD 3-Clause License (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://opensource.org/licenses/BSD-3-Clause
+//
+// Unless required by applicable law or agreed to in writing, software distributed
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations under the License.
+
+use std::sync::Arc;
+
+use crate::core::{
+    context::SDKContext,
+    model::{config::ConfigFile, error::PolarisError},
+};
+
+use super::{
+    api::{ConfigFileAPI, ConfigGroupAPI},
+    req::{
+        DeleteConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest,
+        PublishConfigFileRequest, UpdateConfigFileRequest, WatchConfigFileRequest,
+        WatchConfigFileResponse, WatchConfigGroupRequest, WatchConfigGroupResponse,
+    },
+};
+
+/// DefaultConfigFileAPI
+pub struct DefaultConfigFileAPI {
+    context: Arc<SDKContext>,
+    manage_sdk: bool,
+}
+
+impl DefaultConfigFileAPI {
+    pub fn new(context: Arc<SDKContext>, manage_sdk: bool) -> Self {
+        Self {
+            context,
+            manage_sdk: manage_sdk,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl ConfigFileAPI for DefaultConfigFileAPI {
+    async fn get_config_file(
+        &self,
+        req: GetConfigFileRequest,
+    ) -> Result<ConfigFile, PolarisError> {
+        let engine = self.context.get_engine();
+        engine.get_config_file(req).await
+    }
+
+    async fn update_config_file(
+        &self,
+        _req: UpdateConfigFileRequest,
+    ) -> Result<bool, PolarisError> {
+        Ok(true)
+    }
+
+    async fn delete_config_file(
+        &self,
+        _req: DeleteConfigFileRequest,
+    ) -> Result<bool, PolarisError> {
+        Ok(true)
+    }
+
+    async fn publish_config_file(
+        &self,
+        _req: PublishConfigFileRequest,
+    ) -> Result<bool, PolarisError> {
+        Ok(true)
+    }
+
+    async fn watch_config_file(
+        &self,
+        _req: WatchConfigFileRequest,
+    ) -> Result<WatchConfigFileResponse, PolarisError> {
+        todo!()
+    }
+}
+
+/// DefaultConfigGroupAPI
+pub struct DefaultConfigGroupAPI {}
+
+#[async_trait::async_trait]
+impl ConfigGroupAPI for DefaultConfigGroupAPI {
+    async fn get_publish_config_files(
+        &self,
+        req: GetConfigGroupRequest,
+    ) -> Result<Vec<ConfigFile>, PolarisError> {
+        todo!()
+    }
+
+    async fn watch_publish_config_files(
+        &self,
+        req: WatchConfigGroupRequest,
+    ) -> Result<WatchConfigGroupResponse, PolarisError> {
+        todo!()
+    }
+}

src/config/mod.rs

@@ -13,5 +13,7 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
+pub mod req;
+
 pub mod api;
-pub mod req;
+mod default;

src/config/req.rs

@@ -13,3 +13,55 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
+use std::{collections::HashMap, time::Duration};
+
+use crate::core::model::config::{ConfigFile, ConfigFileRequest};
+
+#[derive(Clone, Debug)]
+pub struct GetConfigFileRequest {
+    pub namespace: String,
+    pub group: String,
+    pub file: String,
+    pub timeout: Duration,
+}
+
+#[derive(Clone, Debug)]
+pub struct CreateConfigFileRequest {
+    pub flow_id: String,
+    pub file: ConfigFile,
+}
+
+impl CreateConfigFileRequest {
+    pub fn to_config_request(&self) -> ConfigFileRequest {
+        let mut flow_id = self.flow_id.clone();
+        if flow_id.is_empty() {
+            flow_id = uuid::Uuid::new_v4().to_string();
+        }
+        ConfigFileRequest {
+            flow_id: flow_id,
+            config_file: self.file.clone(),
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct UpdateConfigFileRequest {}
+
+#[derive(Clone, Debug)]
+pub struct DeleteConfigFileRequest {}
+
+#[derive(Clone, Debug)]
+pub struct PublishConfigFileRequest {}
+
+#[derive(Clone, Debug)]
+pub struct WatchConfigFileRequest {}
+
+pub struct WatchConfigFileResponse {}
+
+#[derive(Clone, Debug)]
+pub struct GetConfigGroupRequest {}
+
+#[derive(Clone, Debug)]
+pub struct WatchConfigGroupRequest {}
+
+pub struct WatchConfigGroupResponse {}

src/core/engine.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use tokio::runtime::{Builder, Runtime};
 
+use crate::config::req::{CreateConfigFileRequest, GetConfigFileRequest};
 use crate::core::config::config::Configuration;
 use crate::core::model::cache::{EventType, ResourceEventKey};
 use crate::core::model::error::PolarisError;
@@ -32,6 +33,7 @@ use crate::discovery::req::{
     ServiceRuleResponse,
 };
 
+use super::model::config::ConfigFile;
 use super::plugin::cache::{Filter, ResourceCache, ResourceListener};
 use super::plugin::connector::Connector;
 use super::plugin::location::{LocationProvider, LocationSupplier};
@@ -267,7 +269,52 @@ impl Engine {
         })
     }
 
-    pub async fn register_resource_listener(&self, listener: Box<dyn ResourceListener>) {
+    /// get_config_file 获取配置文件
+    pub async fn get_config_file(
+        &self,
+        req: GetConfigFileRequest,
+    ) -> Result<ConfigFile, PolarisError> {
+        let local_cache = self.local_cache.clone();
+        let mut filter = HashMap::<String, String>::new();
+        filter.insert("group".to_string(), req.group.clone());
+        filter.insert("file".to_string(), req.file.clone());
+        let ret = local_cache
+            .load_config_file(Filter {
+                resource_key: ResourceEventKey {
+                    namespace: req.namespace.clone(),
+                    event_type: EventType::ConfigFile,
+                    filter,
+                },
+                internal_request: false,
+                include_cache: true,
+                timeout: req.timeout,
+            })
+            .await;
+
+        if ret.is_err() {
+            return Err(ret.err().unwrap());
+        }
+
+        Ok(ret.unwrap())
+    }
+
+    /// create_config_file 创建配置文件
+    pub async fn create_config_file(
+        &self,
+        req: CreateConfigFileRequest,
+    ) -> Result<bool, PolarisError> {
+        let config_file = req.to_config_request();
+
+        let connector = self.server_connector.clone();
+        let rsp = connector.create_config_file(config_file).await;
+
+        return match rsp {
+            Ok(ret_rsp) => Ok(ret_rsp),
+            Err(err) => Err(err),
+        };
+    }
+
+    pub async fn register_resource_listener(&self, listener: Arc<dyn ResourceListener>) {
         self.local_cache.register_resource_listener(listener).await;
     }