You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2023/09/01 08:17:40 UTC
[dubbo-rust] branch feat/cluster updated: feat: Add Router Module(#144) (#153)
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch feat/cluster
in repository https://gitbox.apache.org/repos/asf/dubbo-rust.git
The following commit(s) were added to refs/heads/feat/cluster by this push:
new c07b96d feat: Add Router Module(#144) (#153)
c07b96d is described below
commit c07b96d4ca8300699657b97d18ba788b81417d5c
Author: Urara <95...@users.noreply.github.com>
AuthorDate: Fri Sep 1 16:17:36 2023 +0800
feat: Add Router Module(#144) (#153)
add condition router,
add tag router,
use nacos as router config center
---
application.yaml | 35 +++-
config/src/config.rs | 6 +-
config/src/lib.rs | 1 +
config/src/router.rs | 67 ++++++
dubbo/Cargo.toml | 5 +
dubbo/src/cluster/mod.rs | 28 +--
.../cluster/router/condition/condition_router.rs | 60 ++++++
dubbo/src/cluster/router/condition/matcher.rs | 94 +++++++++
dubbo/src/cluster/router/condition/mod.rs | 3 +
.../src/cluster/router/condition/single_router.rs | 225 +++++++++++++++++++++
.../cluster/router/manager/condition_manager.rs | 75 +++++++
dubbo/src/cluster/router/manager/mod.rs | 3 +
dubbo/src/cluster/router/manager/router_manager.rs | 161 +++++++++++++++
dubbo/src/cluster/router/manager/tag_manager.rs | 27 +++
dubbo/src/cluster/router/mod.rs | 25 +++
.../src/cluster/router/nacos_config_center/mod.rs | 1 +
.../router/nacos_config_center/nacos_client.rs | 132 ++++++++++++
dubbo/src/cluster/router/router_chain.rs | 74 +++++++
dubbo/src/cluster/router/tag/mod.rs | 1 +
dubbo/src/cluster/router/tag/tag_router.rs | 70 +++++++
dubbo/src/cluster/router/utils.rs | 16 ++
dubbo/src/triple/client/builder.rs | 3 +-
22 files changed, 1084 insertions(+), 28 deletions(-)
diff --git a/application.yaml b/application.yaml
index d357db1..902b0ef 100644
--- a/application.yaml
+++ b/application.yaml
@@ -21,4 +21,37 @@ dubbo:
references:
GreeterClientImpl:
url: tri://localhost:20000
- protocol: tri
\ No newline at end of file
+ protocol: tri
+ routers:
+ consumer:
+ - service: "org.apache.dubbo.sample.tri.Greeter"
+ url: triple://localhost:20000
+ protocol: triple
+ nacos:
+ addr: "127.0.0.1:8848"
+ namespace: ""
+ app: ""
+ conditions:
+ - scope: "service"
+ force: false
+ runtime: true
+ enabled: true
+ key: "org.apache.dubbo.sample.tri.Greeter"
+ conditions:
+ - method=greet => port=8889
+ - scope: "service"
+ force: true
+ runtime: true
+ enabled: true
+ key: "user.UserService"
+ conditions:
+ - method=get_s => port=2003
+ tags:
+ force: true
+ enabled: true
+ key: shop-detail
+ tags:
+ - name: gray
+ matches:
+ - key: env
+ value: gray
\ No newline at end of file
diff --git a/config/src/config.rs b/config/src/config.rs
index 646873d..cf4e441 100644
--- a/config/src/config.rs
+++ b/config/src/config.rs
@@ -17,7 +17,7 @@
use std::{collections::HashMap, env, path::PathBuf};
-use crate::{protocol::Protocol, registry::RegistryConfig};
+use crate::{protocol::Protocol, registry::RegistryConfig, router::RouterConfig};
use dubbo_logger::tracing;
use dubbo_utils::yaml_util::yaml_file_parser;
use once_cell::sync::OnceCell;
@@ -44,6 +44,9 @@ pub struct RootConfig {
#[serde(default)]
pub registries: HashMap<String, RegistryConfig>,
+ #[serde(default)]
+ pub routers: RouterConfig,
+
#[serde(default)]
pub data: HashMap<String, String>,
}
@@ -63,6 +66,7 @@ impl RootConfig {
protocols: HashMap::new(),
registries: HashMap::new(),
provider: ProviderConfig::new(),
+ routers: RouterConfig::default(),
data: HashMap::new(),
}
}
diff --git a/config/src/lib.rs b/config/src/lib.rs
index 0748c66..6fa3880 100644
--- a/config/src/lib.rs
+++ b/config/src/lib.rs
@@ -21,4 +21,5 @@ pub mod config;
pub mod protocol;
pub mod provider;
pub mod registry;
+pub mod router;
pub mod service;
diff --git a/config/src/router.rs b/config/src/router.rs
new file mode 100644
index 0000000..98aa14a
--- /dev/null
+++ b/config/src/router.rs
@@ -0,0 +1,67 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
+pub struct ConditionRouterConfig {
+ pub scope: String,
+ pub force: bool,
+ pub runtime: bool,
+ pub enabled: bool,
+ pub key: String,
+ pub conditions: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)]
+pub struct TagRouterConfig {
+ pub force: bool,
+ pub enabled: bool,
+ pub key: String,
+ pub tags: Vec<Tag>,
+}
+
+#[derive(Serialize, Deserialize, Clone, PartialEq, Default, Debug)]
+pub struct ConsumerConfig {
+ pub service: String,
+ pub url: String,
+ pub protocol: String,
+}
+
+#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)]
+pub struct Tag {
+ pub name: String,
+ pub matches: Vec<TagMatchRule>,
+}
+
+#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)]
+pub struct TagMatchRule {
+ pub key: String,
+ pub value: String,
+}
+
+impl ConditionRouterConfig {
+ pub fn new(config: &String) -> Self {
+ serde_yaml::from_str(config).expect("parse error")
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq)]
+pub struct EnableAuth {
+ pub auth_username: String,
+ pub auth_password: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq)]
+pub struct NacosConfig {
+ pub addr: String,
+ pub namespace: String,
+ pub app: String,
+ pub enable_auth: Option<EnableAuth>,
+ pub enable_auth_plugin_http: Option<bool>,
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
+pub struct RouterConfig {
+ pub consumer: Option<Vec<ConsumerConfig>>,
+ pub nacos: Option<NacosConfig>,
+ pub conditions: Option<Vec<ConditionRouterConfig>>,
+ pub tags: Option<TagRouterConfig>,
+}
diff --git a/dubbo/Cargo.toml b/dubbo/Cargo.toml
index 51ccc19..736f0e8 100644
--- a/dubbo/Cargo.toml
+++ b/dubbo/Cargo.toml
@@ -40,8 +40,13 @@ urlencoding.workspace = true
lazy_static.workspace = true
dubbo-base.workspace = true
dubbo-logger.workspace = true
+once_cell.workspace = true
dubbo-config = { path = "../config", version = "0.3.0" }
#对象存储
state = { version = "0.5", features = ["tls"] }
+
+regex = "1.9.1"
+nacos-sdk = { version = "0.3.0", features = ["default"] }
+serde_yaml = "0.9.22"
\ No newline at end of file
diff --git a/dubbo/src/cluster/mod.rs b/dubbo/src/cluster/mod.rs
index afc3cc9..e6a5ffa 100644
--- a/dubbo/src/cluster/mod.rs
+++ b/dubbo/src/cluster/mod.rs
@@ -30,6 +30,7 @@ use crate::{
pub mod directory;
pub mod loadbalance;
+pub mod router;
pub trait Directory: Debug {
fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<BoxInvoker>;
@@ -116,8 +117,9 @@ pub struct MockDirectory {
impl MockDirectory {
pub fn new() -> MockDirectory {
+ // let router_chain = get_global_router_manager().read().unwrap().get_router_chain(invocation);
Self {
- // router_chain: RouterChain::default(),
+ // router_chain
}
}
}
@@ -130,27 +132,3 @@ impl Directory for MockDirectory {
// self.router_chain.route(u, invo);
}
}
-
-// #[derive(Debug, Default)]
-// pub struct RouterChain {
-// router: HashMap<String, BoxRouter>,
-// invokers: Arc<Vec<BoxInvoker>>,
-// }
-
-// impl RouterChain {
-// pub fn route(&mut self, url: Url, invo: Arc<RpcInvocation>) -> Arc<Vec<BoxInvoker>> {
-// let r = self.router.get("mock").unwrap();
-// r.route(self.invokers.clone(), url, invo)
-// }
-// }
-
-// pub trait Router: Debug {
-// fn route(
-// &self,
-// invokers: Arc<Vec<BoxInvoker>>,
-// url: Url,
-// invo: Arc<RpcInvocation>,
-// ) -> Arc<Vec<BoxInvoker>>;
-// }
-
-// pub type BoxRouter = Box<dyn Router + Sync + Send>;
diff --git a/dubbo/src/cluster/router/condition/condition_router.rs b/dubbo/src/cluster/router/condition/condition_router.rs
new file mode 100644
index 0000000..f39cd1c
--- /dev/null
+++ b/dubbo/src/cluster/router/condition/condition_router.rs
@@ -0,0 +1,60 @@
+use crate::{
+ cluster::router::{condition::single_router::ConditionSingleRouter, Router},
+ codegen::RpcInvocation,
+};
+use dubbo_base::Url;
+use std::{
+ fmt::Debug,
+ sync::{Arc, RwLock},
+};
+
+#[derive(Default, Debug, Clone)]
+pub struct ConditionRouter {
+ //condition router for service scope
+ pub service_routers: Option<Arc<RwLock<ConditionSingleRouters>>>,
+ //condition router for application scope
+ pub application_routers: Option<Arc<RwLock<ConditionSingleRouters>>>,
+}
+
+impl Router for ConditionRouter {
+ fn route(&self, invokers: Vec<Url>, url: Url, invo: Arc<RpcInvocation>) -> Vec<Url> {
+ let mut invokers_result = invokers.clone();
+ if let Some(routers) = self.application_routers.clone() {
+ for router in &routers.read().unwrap().routers {
+ invokers_result = router.route(invokers_result, url.clone(), invo.clone())
+ }
+ }
+ if let Some(routers) = self.service_routers.clone() {
+ for router in &routers.read().unwrap().routers {
+ invokers_result = router.route(invokers_result, url.clone(), invo.clone())
+ }
+ }
+ invokers_result
+ }
+}
+
+impl ConditionRouter {
+ pub fn new(
+ service_routers: Option<Arc<RwLock<ConditionSingleRouters>>>,
+ application_routers: Option<Arc<RwLock<ConditionSingleRouters>>>,
+ ) -> Self {
+ Self {
+ service_routers,
+ application_routers,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct ConditionSingleRouters {
+ pub routers: Vec<ConditionSingleRouter>,
+}
+
+impl ConditionSingleRouters {
+ pub fn new(routers: Vec<ConditionSingleRouter>) -> Self {
+ Self { routers }
+ }
+ pub fn is_null(&self) -> bool {
+ self.routers.is_empty()
+ }
+}
diff --git a/dubbo/src/cluster/router/condition/matcher.rs b/dubbo/src/cluster/router/condition/matcher.rs
new file mode 100644
index 0000000..8c9177e
--- /dev/null
+++ b/dubbo/src/cluster/router/condition/matcher.rs
@@ -0,0 +1,94 @@
+use crate::codegen::RpcInvocation;
+use regex::Regex;
+use std::{collections::HashSet, error::Error, option::Option, sync::Arc};
+
+#[derive(Clone, Debug, Default)]
+pub struct ConditionMatcher {
+ _key: String,
+ matches: HashSet<String>,
+ mismatches: HashSet<String>,
+}
+
+impl ConditionMatcher {
+ pub fn new(_key: String) -> Self {
+ ConditionMatcher {
+ _key,
+ matches: HashSet::new(),
+ mismatches: HashSet::new(),
+ }
+ }
+
+ pub fn is_match(
+ &self,
+ value: Option<String>,
+ invocation: Arc<RpcInvocation>,
+ _is_when: bool,
+ ) -> Result<bool, Box<dyn Error>> {
+ match value {
+ None => {
+ // if key does not present in whichever of url, invocation or attachment based on the matcher type, then return false.
+ Ok(false)
+ }
+ Some(val) => {
+ if !self.matches.is_empty() && self.mismatches.is_empty() {
+ for match_ in self.matches.iter() {
+ if self.do_pattern_match(match_, &val, invocation.clone())? {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+ } else if !self.mismatches.is_empty() && self.matches.is_empty() {
+ for mismatch in self.mismatches.iter() {
+ if self.do_pattern_match(mismatch, &val, invocation.clone())? {
+ return Ok(false);
+ }
+ }
+ Ok(true)
+ } else if !self.matches.is_empty() && !self.mismatches.is_empty() {
+ for mismatch in self.mismatches.iter() {
+ if self.do_pattern_match(mismatch, &val, invocation.clone())? {
+ return Ok(false);
+ }
+ }
+ for match_ in self.matches.iter() {
+ if self.do_pattern_match(match_, &val, invocation.clone())? {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+ } else {
+ Ok(false)
+ }
+ }
+ }
+ }
+ pub fn get_matches(&mut self) -> &mut HashSet<String> {
+ &mut self.matches
+ }
+ pub fn get_mismatches(&mut self) -> &mut HashSet<String> {
+ &mut self.mismatches
+ }
+
+ fn do_pattern_match(
+ &self,
+ pattern: &String,
+ value: &String,
+ _invocation: Arc<RpcInvocation>,
+ ) -> Result<bool, Box<dyn Error>> {
+ if pattern.contains("*") {
+ return Ok(star_matcher(pattern, value));
+ }
+ Ok(pattern.eq(value))
+ }
+}
+
+pub fn star_matcher(pattern: &str, input: &str) -> bool {
+ // 将*替换为任意字符的正则表达式
+ let pattern = pattern.replace("*", ".*");
+ let regex = Regex::new(&pattern).unwrap();
+ regex.is_match(input)
+}
+
+pub fn _range_matcher(val: i32, min: i32, max: i32) -> bool {
+ min <= val && val <= max
+}
diff --git a/dubbo/src/cluster/router/condition/mod.rs b/dubbo/src/cluster/router/condition/mod.rs
new file mode 100644
index 0000000..7285b88
--- /dev/null
+++ b/dubbo/src/cluster/router/condition/mod.rs
@@ -0,0 +1,3 @@
+pub mod condition_router;
+pub mod matcher;
+pub mod single_router;
diff --git a/dubbo/src/cluster/router/condition/single_router.rs b/dubbo/src/cluster/router/condition/single_router.rs
new file mode 100644
index 0000000..e290b15
--- /dev/null
+++ b/dubbo/src/cluster/router/condition/single_router.rs
@@ -0,0 +1,225 @@
+use dubbo_base::Url;
+use dubbo_logger::tracing::info;
+use regex::Regex;
+use std::{
+ collections::HashMap,
+ sync::{Arc, RwLock},
+};
+
+use crate::{
+ cluster::router::{condition::matcher::ConditionMatcher, utils::to_original_map, Router},
+ codegen::RpcInvocation,
+ invocation::Invocation,
+};
+
+#[derive(Debug, Clone, Default)]
+pub struct ConditionSingleRouter {
+ pub name: String,
+ pub when_condition: HashMap<String, Arc<RwLock<ConditionMatcher>>>,
+ pub then_condition: HashMap<String, Arc<RwLock<ConditionMatcher>>>,
+ pub enabled: bool,
+ pub force: bool,
+}
+
+impl Router for ConditionSingleRouter {
+ fn route(&self, invokers: Vec<Url>, url: Url, invocation: Arc<RpcInvocation>) -> Vec<Url> {
+ if !self.enabled {
+ return invokers;
+ };
+ let mut result = Vec::new();
+ if self.match_when(url.clone(), invocation.clone()) {
+ for invoker in &invokers.clone() {
+ if self.match_then(invoker.clone(), invocation.clone()) {
+ result.push(invoker.clone());
+ }
+ }
+ if result.is_empty() && self.force == false {
+ invokers
+ } else {
+ result
+ }
+ } else {
+ invokers
+ }
+ }
+}
+
+impl ConditionSingleRouter {
+ pub fn new(rule: String, force: bool, enabled: bool) -> Self {
+ let mut router = Self {
+ name: "condition".to_string(),
+ when_condition: HashMap::new(),
+ then_condition: HashMap::new(),
+ enabled,
+ force,
+ };
+ if enabled {
+ router.init(rule).expect("parse rule error");
+ }
+ router
+ }
+
+ fn init(&mut self, rule: String) -> Result<(), Box<dyn std::error::Error>> {
+ match rule.trim().is_empty() {
+ true => Err("Illegal route rule!".into()),
+ false => {
+ let r = rule.replace("consumer.", "").replace("provider.", "");
+ let i = r.find("=>").unwrap_or_else(|| r.len());
+ let when_rule = r[..i].trim().to_string();
+ let then_rule = r[(i + 2)..].trim().to_string();
+ let when = if when_rule.is_empty() || when_rule == "true" {
+ HashMap::new()
+ } else {
+ self.parse_rule(&when_rule)?
+ };
+ let then = if then_rule.is_empty() || then_rule == "false" {
+ HashMap::new()
+ } else {
+ self.parse_rule(&then_rule)?
+ };
+ self.when_condition = when;
+ self.then_condition = then;
+ Ok(())
+ }
+ }
+ }
+
+ fn parse_rule(
+ &mut self,
+ rule: &str,
+ ) -> Result<HashMap<String, Arc<RwLock<ConditionMatcher>>>, Box<dyn std::error::Error>> {
+ let mut conditions: HashMap<String, Arc<RwLock<ConditionMatcher>>> = HashMap::new();
+ let mut current_matcher: Option<Arc<RwLock<ConditionMatcher>>> = None;
+ let regex = Regex::new(r"([&!=,]*)\s*([^&!=,\s]+)").unwrap();
+ for cap in regex.captures_iter(rule) {
+ let separator = &cap[1];
+ let content = &cap[2];
+
+ match separator {
+ "" => {
+ let current_key = content.to_string();
+ current_matcher =
+ Some(Arc::new(RwLock::new(self.get_matcher(current_key.clone()))));
+ conditions.insert(
+ current_key.clone(),
+ current_matcher.as_ref().unwrap().clone(),
+ );
+ }
+ "&" => {
+ let current_key = content.to_string();
+ current_matcher = conditions.get(¤t_key).cloned();
+ if current_matcher.is_none() {
+ let matcher = Arc::new(RwLock::new(self.get_matcher(current_key.clone())));
+ conditions.insert(current_key.clone(), matcher.clone());
+ current_matcher = Some(matcher);
+ }
+ }
+ "=" => {
+ if let Some(matcher) = ¤t_matcher {
+ let mut matcher_borrowed = matcher.write().unwrap();
+ matcher_borrowed
+ .get_matches()
+ .insert(content.to_string().clone());
+ } else {
+ return Err("Error: ...".into());
+ }
+ }
+ "!=" => {
+ if let Some(matcher) = ¤t_matcher {
+ let mut matcher_borrowed = matcher.write().unwrap();
+ matcher_borrowed
+ .get_mismatches()
+ .insert(content.to_string().clone());
+ } else {
+ return Err("Error: ...".into());
+ }
+ }
+ "," => {
+ if let Some(matcher) = ¤t_matcher {
+ let mut matcher_borrowed = matcher.write().unwrap();
+ if matcher_borrowed.get_matches().is_empty()
+ && matcher_borrowed.get_mismatches().is_empty()
+ {
+ return Err("Error: ...".into());
+ }
+ drop(matcher_borrowed);
+ let mut matcher_borrowed_mut = matcher.write().unwrap();
+ matcher_borrowed_mut
+ .get_matches()
+ .insert(content.to_string().clone());
+ } else {
+ return Err("Error: ...".into());
+ }
+ }
+ _ => {
+ return Err("Error: ...".into());
+ }
+ }
+ }
+ Ok(conditions)
+ }
+
+ // pub fn is_runtime(&self) -> bool {
+ // // same as the Java version
+ // }
+
+ pub fn get_matcher(&self, key: String) -> ConditionMatcher {
+ ConditionMatcher::new(key)
+ }
+
+ pub fn match_when(&self, url: Url, invocation: Arc<RpcInvocation>) -> bool {
+ if self.when_condition.is_empty() {
+ true
+ } else {
+ false
+ };
+ self.do_match(url, &self.when_condition, invocation, true)
+ }
+
+ pub fn match_then(&self, url: Url, invocation: Arc<RpcInvocation>) -> bool {
+ if self.when_condition.is_empty() {
+ true
+ } else {
+ false
+ };
+ self.do_match(url, &self.then_condition, invocation, false)
+ }
+
+ pub fn do_match(
+ &self,
+ url: Url,
+ conditions: &HashMap<String, Arc<RwLock<ConditionMatcher>>>,
+ invocation: Arc<RpcInvocation>,
+ is_when: bool,
+ ) -> bool {
+ let sample: HashMap<String, String> = to_original_map(url);
+ for (key, condition_matcher) in conditions {
+ let matcher = condition_matcher.read().unwrap();
+ let value = get_value(key, &sample, invocation.clone());
+ match matcher.is_match(value, invocation.clone(), is_when) {
+ Ok(result) => {
+ if !result {
+ return false;
+ }
+ }
+ Err(error) => {
+ info!("Error occurred: {:?}", error);
+ return false;
+ }
+ }
+ }
+ true
+ }
+}
+
+fn get_value(
+ key: &String,
+ sample: &HashMap<String, String>,
+ invocation: Arc<RpcInvocation>,
+) -> Option<String> {
+ if key == "method" {
+ let method_param = invocation.get_method_name();
+ return Some(method_param);
+ }
+ sample.get(key).cloned()
+}
diff --git a/dubbo/src/cluster/router/manager/condition_manager.rs b/dubbo/src/cluster/router/manager/condition_manager.rs
new file mode 100644
index 0000000..4207b79
--- /dev/null
+++ b/dubbo/src/cluster/router/manager/condition_manager.rs
@@ -0,0 +1,75 @@
+use crate::cluster::router::condition::{
+ condition_router::{ConditionRouter, ConditionSingleRouters},
+ single_router::ConditionSingleRouter,
+};
+use dubbo_config::router::ConditionRouterConfig;
+use std::{
+ collections::HashMap,
+ sync::{Arc, RwLock},
+};
+
+#[derive(Debug, Clone, Default)]
+pub struct ConditionRouterManager {
+ //Application-level routing applies globally, while service-level routing only affects a specific service.
+ pub routers_service: HashMap<String, Arc<RwLock<ConditionSingleRouters>>>,
+ pub routers_application: Arc<RwLock<ConditionSingleRouters>>,
+}
+
+impl ConditionRouterManager {
+ pub fn get_router(&self, service_name: String) -> Option<ConditionRouter> {
+ let routers_service = self.routers_service.get(&service_name);
+ match routers_service {
+ Some(routers_service) => {
+ if self.routers_application.read().unwrap().is_null() {
+ return Some(ConditionRouter::new(Some(routers_service.clone()), None));
+ }
+ Some(ConditionRouter::new(
+ Some(routers_service.clone()),
+ Some(self.routers_application.clone()),
+ ))
+ }
+ None => {
+ if self.routers_application.read().unwrap().is_null() {
+ return None;
+ }
+ Some(ConditionRouter::new(
+ None,
+ Some(self.routers_application.clone()),
+ ))
+ }
+ }
+ }
+
+ pub fn update(&mut self, config: ConditionRouterConfig) {
+ let force = config.force;
+ let scope = config.scope;
+ let key = config.key;
+ let enable = config.enabled;
+ let mut routers = Vec::new();
+ for condition in config.conditions {
+ routers.push(ConditionSingleRouter::new(condition, force, enable));
+ }
+ match scope.as_str() {
+ "application" => {
+ self.routers_application.write().unwrap().routers = routers;
+ }
+ "service" => {
+ if let Some(x) = self.routers_service.get(&key) {
+ x.write().unwrap().routers = routers
+ } else {
+ self.routers_service.insert(
+ key,
+ Arc::new(RwLock::new(ConditionSingleRouters::new(routers))),
+ );
+ }
+ }
+ _ => {}
+ }
+ }
+
+ pub fn _parse_rules(&mut self, configs: Vec<ConditionRouterConfig>) {
+ for config in configs {
+ self.update(config)
+ }
+ }
+}
diff --git a/dubbo/src/cluster/router/manager/mod.rs b/dubbo/src/cluster/router/manager/mod.rs
new file mode 100644
index 0000000..025f6c1
--- /dev/null
+++ b/dubbo/src/cluster/router/manager/mod.rs
@@ -0,0 +1,3 @@
+mod condition_manager;
+pub mod router_manager;
+mod tag_manager;
diff --git a/dubbo/src/cluster/router/manager/router_manager.rs b/dubbo/src/cluster/router/manager/router_manager.rs
new file mode 100644
index 0000000..a2a6658
--- /dev/null
+++ b/dubbo/src/cluster/router/manager/router_manager.rs
@@ -0,0 +1,161 @@
+use crate::{
+ cluster::router::{
+ manager::{condition_manager::ConditionRouterManager, tag_manager::TagRouterManager},
+ nacos_config_center::nacos_client::NacosClient,
+ router_chain::RouterChain,
+ },
+ invocation::{Invocation, RpcInvocation},
+};
+use dubbo_base::Url;
+use dubbo_config::{
+ get_global_config,
+ router::{ConditionRouterConfig, NacosConfig, TagRouterConfig},
+};
+use dubbo_logger::tracing::{info, trace};
+use once_cell::sync::OnceCell;
+use std::{
+ collections::HashMap,
+ sync::{Arc, RwLock},
+};
+
+pub static GLOBAL_ROUTER_MANAGER: OnceCell<Arc<RwLock<RouterManager>>> = OnceCell::new();
+
+pub struct RouterManager {
+ pub condition_router_manager: ConditionRouterManager,
+ pub tag_router_manager: TagRouterManager,
+ pub nacos: Option<NacosClient>,
+ pub consumer: HashMap<String, Url>,
+}
+
+impl RouterManager {
+ pub fn get_router_chain(&self, invocation: Arc<RpcInvocation>) -> RouterChain {
+ let service = invocation.get_target_service_unique_name().clone();
+ let condition_router = self.condition_router_manager.get_router(service.clone());
+ let tag_router = self.tag_router_manager.get_router();
+ let mut chain = RouterChain::new();
+ match self.consumer.get(service.as_str()) {
+ None => {}
+ Some(url) => {
+ chain.set_condition_router(condition_router);
+ chain.set_tag_router(tag_router);
+ chain.self_url = url.clone();
+ }
+ }
+ chain
+ }
+
+ pub fn notify(&mut self, event: RouterConfigChangeEvent) {
+ match event.router_kind.as_str() {
+ "condition" => {
+ let config: ConditionRouterConfig =
+ serde_yaml::from_str(event.content.as_str()).unwrap();
+ self.condition_router_manager.update(config)
+ }
+ "tag" => {
+ let config: TagRouterConfig = serde_yaml::from_str(event.content.as_str()).unwrap();
+ self.tag_router_manager.update(config)
+ }
+ _ => {
+ info!("other router change event")
+ }
+ }
+ }
+
+ pub fn init_nacos(&mut self, config: NacosConfig) {
+ self.nacos = Some(NacosClient::new_init_client(config));
+ self.init_router_managers_for_nacos();
+ }
+
+ pub fn init_router_managers_for_nacos(&mut self) {
+ let config = self
+ .nacos
+ .as_ref()
+ .unwrap()
+ .get_tag_config("application".to_string());
+ match config {
+ None => {}
+ Some(tag_config) => {
+ self.tag_router_manager.init();
+ self.tag_router_manager.update(tag_config)
+ }
+ }
+ for (service_name, _) in &self.consumer {
+ let config = self
+ .nacos
+ .as_ref()
+ .unwrap()
+ .get_condition_config(service_name.clone());
+ match config {
+ None => {}
+ Some(condition_config) => self.condition_router_manager.update(condition_config),
+ }
+ }
+ }
+
+ pub fn init(&mut self) {
+ let config = get_global_config().routers.clone();
+ let consumer_configs = get_global_config()
+ .routers
+ .consumer
+ .clone()
+ .unwrap_or(Vec::new());
+ for consumer_config in consumer_configs {
+ self.consumer.insert(
+ consumer_config.service.clone(),
+ Url::from_url(
+ format!("{}/{}", consumer_config.url, consumer_config.service).as_str(),
+ )
+ .expect("consumer配置出错！Url生成错误"),
+ );
+ }
+ match &config.nacos {
+ None => {
+ trace!("Nacos not configured, using local YAML configuration for routing");
+ let condition = config.conditions.clone();
+ match condition {
+ None => {
+ info!("Unconfigured Condition Router")
+ }
+ Some(cons) => {
+ for con in cons {
+ self.condition_router_manager.update(con)
+ }
+ }
+ }
+ let tag = config.tags.clone();
+ match tag {
+ None => {
+ info!("Unconfigured Tag Router")
+ }
+ Some(ta) => {
+ self.tag_router_manager.init();
+ self.tag_router_manager.update(ta)
+ }
+ }
+ }
+ Some(config) => {
+ self.init_nacos(config.clone());
+ }
+ }
+ }
+}
+
+pub fn get_global_router_manager() -> &'static Arc<RwLock<RouterManager>> {
+ GLOBAL_ROUTER_MANAGER.get_or_init(|| {
+ let mut router_manager = RouterManager {
+ condition_router_manager: ConditionRouterManager::default(),
+ tag_router_manager: TagRouterManager::default(),
+ nacos: None,
+ consumer: HashMap::new(),
+ };
+ router_manager.init();
+ return Arc::new(RwLock::new(router_manager));
+ })
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct RouterConfigChangeEvent {
+ pub service_name: String,
+ pub router_kind: String,
+ pub content: String,
+}
diff --git a/dubbo/src/cluster/router/manager/tag_manager.rs b/dubbo/src/cluster/router/manager/tag_manager.rs
new file mode 100644
index 0000000..ce30c92
--- /dev/null
+++ b/dubbo/src/cluster/router/manager/tag_manager.rs
@@ -0,0 +1,27 @@
+use crate::cluster::router::tag::tag_router::TagRouter;
+use dubbo_config::router::TagRouterConfig;
+use std::sync::{Arc, RwLock};
+
+#[derive(Debug, Clone, Default)]
+pub struct TagRouterManager {
+ pub tag_router: Option<Arc<RwLock<TagRouter>>>,
+}
+
+impl TagRouterManager {
+ pub fn init(&mut self) {
+ self.tag_router = Some(Arc::new(RwLock::new(TagRouter::default())))
+ }
+
+ pub fn get_router(&self) -> Option<Arc<RwLock<TagRouter>>> {
+ self.tag_router.clone()
+ }
+
+ pub fn update(&mut self, config: TagRouterConfig) {
+ self.tag_router
+ .as_ref()
+ .unwrap()
+ .write()
+ .unwrap()
+ .parse_config(config)
+ }
+}
diff --git a/dubbo/src/cluster/router/mod.rs b/dubbo/src/cluster/router/mod.rs
new file mode 100644
index 0000000..84ceae1
--- /dev/null
+++ b/dubbo/src/cluster/router/mod.rs
@@ -0,0 +1,25 @@
+pub mod condition;
+pub mod manager;
+pub mod nacos_config_center;
+pub mod router_chain;
+pub mod tag;
+pub mod utils;
+
+use crate::invocation::RpcInvocation;
+use dubbo_base::Url;
+use std::{fmt::Debug, sync::Arc};
+
+pub trait Router: Debug + Clone {
+ fn route(&self, invokers: Vec<Url>, url: Url, invo: Arc<RpcInvocation>) -> Vec<Url>;
+}
+
+// pub type BoxRouter = Box<dyn Router + Sync + Send>;
+
+#[derive(Debug, Default, Clone)]
+pub struct MockRouter {}
+
+impl Router for MockRouter {
+ fn route(&self, invokers: Vec<Url>, _url: Url, _invo: Arc<RpcInvocation>) -> Vec<Url> {
+ invokers
+ }
+}
diff --git a/dubbo/src/cluster/router/nacos_config_center/mod.rs b/dubbo/src/cluster/router/nacos_config_center/mod.rs
new file mode 100644
index 0000000..7878fa9
--- /dev/null
+++ b/dubbo/src/cluster/router/nacos_config_center/mod.rs
@@ -0,0 +1 @@
+pub mod nacos_client;
diff --git a/dubbo/src/cluster/router/nacos_config_center/nacos_client.rs b/dubbo/src/cluster/router/nacos_config_center/nacos_client.rs
new file mode 100644
index 0000000..1600e9b
--- /dev/null
+++ b/dubbo/src/cluster/router/nacos_config_center/nacos_client.rs
@@ -0,0 +1,132 @@
+use crate::cluster::router::manager::router_manager::{
+ get_global_router_manager, RouterConfigChangeEvent,
+};
+use dubbo_config::router::{ConditionRouterConfig, NacosConfig, TagRouterConfig};
+use dubbo_logger::{tracing, tracing::info};
+use nacos_sdk::api::{
+ config::{ConfigChangeListener, ConfigResponse, ConfigService, ConfigServiceBuilder},
+ props::ClientProps,
+};
+use std::sync::{Arc, RwLock};
+
+pub struct NacosClient {
+ pub client: Arc<RwLock<dyn ConfigService>>,
+}
+
+unsafe impl Send for NacosClient {}
+
+unsafe impl Sync for NacosClient {}
+
+pub struct ConfigChangeListenerImpl;
+
+impl NacosClient {
+ pub fn new_init_client(config: NacosConfig) -> Self {
+ let server_addr = config.addr;
+ let namespace = config.namespace;
+ let app = config.app;
+ match config.enable_auth {
+ None => {
+ info!("disable nacos auth!");
+ info!("nacos init,addr:{}", server_addr);
+ let client = Arc::new(RwLock::new(
+ ConfigServiceBuilder::new(
+ ClientProps::new()
+ .server_addr(server_addr)
+ .namespace(namespace)
+ .app_name(app),
+ )
+ .build()
+ .expect("NacosClient build failed!Please check NacosConfig"),
+ ));
+ Self { client }
+ }
+ Some(auth) => {
+ info!("enable nacos auth!");
+ info!("nacos init,addr:{}", server_addr);
+ let client = Arc::new(RwLock::new(
+ ConfigServiceBuilder::new(
+ ClientProps::new()
+ .server_addr(server_addr)
+ .namespace(namespace)
+ .app_name(app)
+ .auth_username(auth.auth_username)
+ .auth_password(auth.auth_password),
+ )
+ // .enable_auth_plugin_http()
+ .build()
+ .expect("NacosClient build failed!Please check NacosConfig"),
+ ));
+ return Self { client };
+ }
+ }
+ }
+ pub fn get_condition_config(&self, data_id: String) -> Option<ConditionRouterConfig> {
+ let config_resp = self
+ .client
+ .read()
+ .unwrap()
+ .get_config(data_id.clone(), "condition".to_string());
+ return match config_resp {
+ Ok(config_resp) => {
+ self.add_listener(data_id.clone(), "condition".to_string());
+ let string = config_resp.content();
+ Some(serde_yaml::from_str(string).unwrap())
+ }
+ Err(_err) => None,
+ };
+ }
+
+ pub fn get_tag_config(&self, data_id: String) -> Option<TagRouterConfig> {
+ let config_resp = self
+ .client
+ .read()
+ .unwrap()
+ .get_config(data_id.clone(), "tag".to_string());
+ return match config_resp {
+ Ok(config_resp) => {
+ self.add_listener(data_id.clone(), "tag".to_string());
+ let string = config_resp.content();
+ let result = serde_yaml::from_str(string);
+ match result {
+ Ok(config) => {
+ info!("success to get TagRouter config and parse success");
+ Some(config)
+ }
+ _ => {
+ info!("failed to parse TagRouter rule");
+ None
+ }
+ }
+ }
+ Err(_err) => None,
+ };
+ }
+ pub fn add_listener(&self, data_id: String, group: String) {
+ let res_listener = self
+ .client
+ .write()
+ .expect("failed to create nacos config listener")
+ .add_listener(data_id, group, Arc::new(ConfigChangeListenerImpl {}));
+ match res_listener {
+ Ok(_) => {
+ info!("listening the config success");
+ }
+ Err(err) => tracing::error!("listen config error {:?}", err),
+ }
+ }
+}
+
+impl ConfigChangeListener for ConfigChangeListenerImpl {
+ fn notify(&self, config_resp: ConfigResponse) {
+ let content_type = config_resp.content_type();
+ let event = RouterConfigChangeEvent {
+ service_name: config_resp.data_id().clone(),
+ router_kind: config_resp.group().clone(),
+ content: config_resp.content().clone(),
+ };
+ if content_type == "yaml" {
+ get_global_router_manager().write().unwrap().notify(event);
+ }
+ info!("notify config={:?}", config_resp);
+ }
+}
diff --git a/dubbo/src/cluster/router/router_chain.rs b/dubbo/src/cluster/router/router_chain.rs
new file mode 100644
index 0000000..930f0f2
--- /dev/null
+++ b/dubbo/src/cluster/router/router_chain.rs
@@ -0,0 +1,74 @@
+use crate::{
+ cluster::router::{
+ condition::condition_router::ConditionRouter, tag::tag_router::TagRouter, Router,
+ },
+ invocation::RpcInvocation,
+};
+use dubbo_base::Url;
+use std::sync::{Arc, RwLock};
+
+#[derive(Debug, Default, Clone)]
+pub struct RouterChain {
+ pub condition_router: Option<ConditionRouter>,
+ pub tag_router: Option<Arc<RwLock<TagRouter>>>,
+ pub self_url: Url,
+}
+
+impl RouterChain {
+ pub fn new() -> Self {
+ RouterChain {
+ condition_router: None,
+ tag_router: None,
+ self_url: Url::new(),
+ }
+ }
+ pub fn set_condition_router(&mut self, router: Option<ConditionRouter>) {
+ self.condition_router = router;
+ }
+ pub fn set_tag_router(&mut self, router: Option<Arc<RwLock<TagRouter>>>) {
+ self.tag_router = router;
+ }
+ pub fn route(&self, invokers: Vec<Url>, invocation: Arc<RpcInvocation>) -> Vec<Url> {
+ let mut result = invokers.clone();
+ match &self.tag_router {
+ None => {}
+ Some(router) => {
+ result =
+ router
+ .read()
+ .unwrap()
+ .route(result, self.self_url.clone(), invocation.clone())
+ }
+ }
+ match &self.condition_router {
+ Some(router) => {
+ result = router.route(result, self.self_url.clone(), invocation.clone())
+ }
+ None => {}
+ }
+ result
+ }
+}
+
+#[test]
+fn test() {
+ use crate::cluster::router::manager::router_manager::get_global_router_manager;
+
+ let u1 = Url::from_url("triple://127.0.0.1:8888/org.apache.dubbo.sample.tri.Greeter").unwrap();
+ let u2 = Url::from_url("triple://127.0.0.1:8889/org.apache.dubbo.sample.tri.Greeter").unwrap();
+ let u3 = Url::from_url("triple://127.0.0.1:8800/org.apache.dubbo.sample.tri.Greeter").unwrap();
+ let u4 = Url::from_url("triple://127.0.2.1:880/org.apache.dubbo.sample.tri.Greeter").unwrap();
+ let u5 = Url::from_url("triple://127.0.1.1:8888/org.apache.dubbo.sample.tri.Greeter").unwrap();
+ let invos = vec![u1, u2, u3, u4, u5];
+ let invo = Arc::new(
+ RpcInvocation::default()
+ .with_method_name("greet".to_string())
+ .with_service_unique_name("org.apache.dubbo.sample.tri.Greeter".to_string()),
+ );
+ let x = get_global_router_manager()
+ .read()
+ .unwrap()
+ .get_router_chain(invo.clone());
+ let result = x.route(invos, invo.clone());
+ dbg!(result);
+}
diff --git a/dubbo/src/cluster/router/tag/mod.rs b/dubbo/src/cluster/router/tag/mod.rs
new file mode 100644
index 0000000..6ac5b21
--- /dev/null
+++ b/dubbo/src/cluster/router/tag/mod.rs
@@ -0,0 +1 @@
+pub mod tag_router;
diff --git a/dubbo/src/cluster/router/tag/tag_router.rs b/dubbo/src/cluster/router/tag/tag_router.rs
new file mode 100644
index 0000000..d1a962e
--- /dev/null
+++ b/dubbo/src/cluster/router/tag/tag_router.rs
@@ -0,0 +1,70 @@
+use crate::{
+ cluster::router::{utils::to_original_map, Router},
+ codegen::RpcInvocation,
+};
+use dubbo_base::Url;
+use dubbo_config::router::TagRouterConfig;
+use std::{collections::HashMap, fmt::Debug, sync::Arc};
+
+#[derive(Debug, Clone, Default)]
+pub struct TagRouter {
+ pub tag_rules: HashMap<String, HashMap<String, String>>,
+ pub force: bool,
+ pub enabled: bool,
+}
+
+impl TagRouter {
+ pub fn parse_config(&mut self, config: TagRouterConfig) {
+ self.tag_rules = HashMap::new();
+ self.force = config.force;
+ self.enabled = config.enabled;
+ for tag in &config.tags {
+ let mut tags = HashMap::new();
+ for rule in &tag.matches {
+ tags.insert(rule.key.clone(), rule.value.clone());
+ }
+ self.tag_rules.insert(tag.name.clone(), tags);
+ }
+ }
+
+ pub fn match_tag(&self, params: HashMap<String, String>) -> Option<String> {
+ let mut tag_result = None;
+ for (tag, tag_rules) in &self.tag_rules {
+ for (key, value) in tag_rules {
+ match params.get(key.as_str()) {
+ None => {}
+ Some(val) => {
+ if val == value {
+ tag_result = Some(tag.clone())
+ }
+ }
+ }
+ }
+ }
+ tag_result
+ }
+}
+
+impl Router for TagRouter {
+ fn route(&self, invokers: Vec<Url>, url: Url, _invocation: Arc<RpcInvocation>) -> Vec<Url> {
+ if !self.enabled {
+ return invokers;
+ };
+ let self_param = to_original_map(url);
+ let invocation_tag = self.match_tag(self_param);
+ let mut invokers_result = Vec::new();
+ for invoker in &invokers {
+ let invoker_param = to_original_map(invoker.clone());
+ let invoker_tag = self.match_tag(invoker_param);
+ if invoker_tag == invocation_tag {
+ invokers_result.push(invoker.clone())
+ }
+ }
+ if invokers_result.is_empty() {
+ if !self.force {
+ return invokers;
+ }
+ }
+ invokers_result
+ }
+}
diff --git a/dubbo/src/cluster/router/utils.rs b/dubbo/src/cluster/router/utils.rs
new file mode 100644
index 0000000..2ca50fc
--- /dev/null
+++ b/dubbo/src/cluster/router/utils.rs
@@ -0,0 +1,16 @@
+use dubbo_base::Url;
+use std::{collections::HashMap, string::String};
+
+pub fn to_original_map(url: Url) -> HashMap<String, String> {
+ let mut result: HashMap<String, String> = HashMap::new();
+ result.insert("scheme".parse().unwrap(), url.scheme);
+ result.insert("location".parse().unwrap(), url.location);
+ result.insert("ip".parse().unwrap(), url.ip);
+ result.insert("port".parse().unwrap(), url.port);
+ result.insert("service_name".parse().unwrap(), url.service_name);
+ result.insert("service_key".parse().unwrap(), url.service_key);
+ for (key, value) in url.params {
+ result.insert(key, value);
+ }
+ result
+}
diff --git a/dubbo/src/triple/client/builder.rs b/dubbo/src/triple/client/builder.rs
index d68067a..b8118e2 100644
--- a/dubbo/src/triple/client/builder.rs
+++ b/dubbo/src/triple/client/builder.rs
@@ -30,6 +30,7 @@ use dubbo_base::Url;
pub type ClientBoxService =
BoxCloneService<http::Request<SdkBody>, http::Response<crate::BoxBody>, crate::Error>;
+#[allow(dead_code)]
#[derive(Clone, Debug, Default)]
pub struct ClientBuilder {
pub timeout: Option<u64>,
@@ -56,7 +57,7 @@ impl ClientBuilder {
connector: "",
directory: Some(Arc::new(Box::new(StaticDirectory::new(&host)))),
direct: true,
- host: host.clone().to_string(),
+ host: host.to_string(),
}
}