feat: 优化gRPC命名空间处理逻辑，增强默认命名空间兼容性

Author: heqingpan

src/grpc/bistream_conn.rs

@@ -13,6 +13,50 @@ use super::{api_model::ClientDetectionRequest, nacos_proto::Payload};
 type SenderType = tokio::sync::mpsc::Sender<Result<Payload, tonic::Status>>;
 type ReceiverStreamType = tonic::Streaming<Payload>;
 
+#[derive(Debug)]
+pub enum NamespaceType {
+    Empty,
+    Public,
+    Other(Arc<String>),
+    Unknown,
+}
+
+impl NamespaceType {
+    pub fn from_option(namespace: Option<String>) -> Self {
+        if let Some(namespace) = namespace {
+            NamespaceType::get_namespace_type(&namespace)
+        } else {
+            NamespaceType::Unknown
+        }
+    }
+    pub fn get_namespace_type(namespace: &str) -> Self {
+        if namespace.is_empty() {
+            NamespaceType::Empty
+        } else if namespace == "public" {
+            NamespaceType::Public
+        } else {
+            NamespaceType::Other(Arc::new(namespace.to_owned()))
+        }
+    }
+
+    pub fn is_default(&self) -> bool {
+        match self {
+            NamespaceType::Empty => true,
+            NamespaceType::Public => true,
+            _ => false,
+        }
+    }
+
+    pub fn to_str(&self) -> &str {
+        match self {
+            NamespaceType::Empty => "",
+            NamespaceType::Public => "public",
+            NamespaceType::Other(namespace) => namespace.as_str(),
+            NamespaceType::Unknown => "",
+        }
+    }
+}
+
 pub struct BiStreamConn {
     sender: SenderType,
     client_id: Arc<String>,

src/grpc/bistream_manage.rs

@@ -22,7 +22,10 @@ use super::{
 };
 use crate::common::constant::EMPTY_CLIENT_VERSION;
 use crate::common::model::ClientVersion;
+use crate::config::ConfigUtils;
 use crate::grpc::api_model::ConnectionSetupRequest;
+use crate::grpc::bistream_conn::NamespaceType;
+use crate::namespace::DEFAULT_NAMESPACE;
 use actix::prelude::*;
 use bean_factory::{bean, Inject};
 use inner_mem_cache::TimeoutSet;
@@ -31,6 +34,7 @@ pub(crate) struct ConnCacheItem {
     last_active_time: u64,
     conn: Addr<BiStreamConn>,
     pub(crate) client_version: Arc<ClientVersion>,
+    pub(crate) namespace: NamespaceType,
 }
 
 impl ConnCacheItem {
@@ -39,6 +43,7 @@ impl ConnCacheItem {
             last_active_time,
             conn,
             client_version: EMPTY_CLIENT_VERSION.clone(),
+            namespace: NamespaceType::Unknown,
         }
     }
 }
@@ -236,6 +241,7 @@ impl Handler<BiStreamManageCmd> for BiStreamManage {
                                 item.client_version =
                                     Arc::new(ClientVersion::from_string(&client_version));
                             }
+                            item.namespace = NamespaceType::from_option(request.tenant);
                         }
                     }
                     self.active_client(client_id).ok();
@@ -260,7 +266,8 @@ impl Handler<BiStreamManageCmd> for BiStreamManage {
                 return Ok(BiStreamManageResult::ClientInfo(client_version));
             }
             BiStreamManageCmd::NotifyConfig(config_key, client_id_set) => {
-                let request = ConfigChangeNotifyRequest {
+                let tenant = config_key.tenant.clone();
+                let mut request = ConfigChangeNotifyRequest {
                     group: config_key.group,
                     data_id: config_key.data_id,
                     tenant: config_key.tenant,
@@ -270,10 +277,27 @@ impl Handler<BiStreamManageCmd> for BiStreamManage {
                 };
                 let payload = Arc::new(PayloadUtils::build_payload(
                     "ConfigChangeNotifyRequest",
-                    serde_json::to_string(&request).unwrap(),
+                    serde_json::to_string(&request)?,
                 ));
+                let mut other_default_payload = None;
                 for item in &client_id_set {
                     if let Some(item) = self.conn_cache.get(item) {
+                        // 默认命名空间有两个值，需要把通知的命令空间值调整为监听的值，以兼容不同版本的客户端
+                        if item.namespace.is_default() && item.namespace.to_str() != tenant.as_str()
+                        {
+                            if other_default_payload.is_none() {
+                                request.tenant = Arc::new(item.namespace.to_str().to_string());
+                                other_default_payload =
+                                    Some(Arc::new(PayloadUtils::build_payload(
+                                        "ConfigChangeNotifyRequest",
+                                        serde_json::to_string(&request)?,
+                                    )));
+                            }
+                            if let Some(payload) = other_default_payload.as_ref() {
+                                item.conn.do_send(BiStreamSenderCmd::Send(payload.clone()));
+                                continue;
+                            }
+                        }
                         item.conn.do_send(BiStreamSenderCmd::Send(payload.clone()));
                     }
                 }

src/grpc/handler/config_change_batch_listen.rs

@@ -2,7 +2,7 @@
 
 use std::sync::Arc;
 
-use crate::config::ConfigUtils;
+use crate::config::{ConfigUtils, DEFAULT_TENANT};
 use crate::grpc::HandlerResult;
 use crate::{
     common::appdata::AppShareData,
@@ -40,7 +40,11 @@ impl PayloadHandler for ConfigChangeBatchListenRequestHandler {
         let body_vec = request_payload.body.unwrap_or_default().value;
         let request: ConfigBatchListenRequest = serde_json::from_slice(&body_vec)?;
         let mut listener_items = vec![];
+        let mut use_public = false;
         for item in request.config_listen_contexts {
+            if item.tenant == DEFAULT_TENANT {
+                use_public = true;
+            }
             let key = ConfigKey::new(
                 &item.data_id,
                 &item.group,
@@ -65,10 +69,15 @@ impl PayloadHandler for ConfigChangeBatchListenRequestHandler {
                     ConfigResult::ChangeKey(keys) => {
                         response.result_code = SUCCESS_CODE;
                         for key in keys {
+                            let tenant = if key.tenant.is_empty() && use_public {
+                                Arc::new(DEFAULT_TENANT.to_owned())
+                            } else {
+                                key.tenant.clone()
+                            };
                             let obj = ConfigContext {
                                 data_id: key.data_id,
                                 group: key.group,
-                                tenant: key.tenant,
+                                tenant,
                             };
                             response.changed_configs.push(obj);
                         }