Rft(dubbo): use registry_url and service_url to impl registry_protocol plugin

Author: yang20150702

dubbo/src/common/url.rs

@@ -96,7 +96,8 @@ mod tests {
 
     #[test]
     fn test_from_url() {
-        let u1 = Url::from_url("");
+        let u1 = Url::from_url("triple://127.0.0.1:8888/helloworld.Greeter");
+        println!("{:?}", u1.unwrap().get_service_name())
     }
 
     #[test]

dubbo/src/framework.rs

@@ -21,7 +21,6 @@ use std::pin::Pin;
 use futures::future;
 use futures::Future;
 
-use crate::common::consts;
 use crate::common::url::Url;
 use crate::protocol::{BoxExporter, Protocol};
 use crate::registry::protocol::RegistryProtocol;
@@ -33,7 +32,7 @@ use dubbo_config::{get_global_config, RootConfig};
 pub struct Dubbo {
     protocols: HashMap<String, Vec<Url>>,
     registries: HashMap<String, Url>,
-    services: HashMap<String, Vec<Url>>, // protocol:  Url; registry: Url
+    service_registry: HashMap<String, Vec<Url>>, // registry: Urls
     config: Option<RootConfig>,
 }
 
@@ -43,7 +42,7 @@ impl Dubbo {
         Self {
             protocols: HashMap::new(),
             registries: HashMap::new(),
-            services: HashMap::new(),
+            service_registry: HashMap::new(),
             config: None,
         }
     }
@@ -94,24 +93,16 @@ impl Dubbo {
             }
 
             let u = u.unwrap();
-            if self.services.get(consts::PROTOCOL).is_some() {
-                self.services
-                    .get_mut(consts::PROTOCOL)
-                    .unwrap()
-                    .push(u.clone());
-            } else {
-                self.services
-                    .insert(consts::PROTOCOL.to_string(), vec![u.clone()]);
-            }
 
-            if self.services.get(consts::REGISTRY).is_some() {
-                self.services
-                    .get_mut(consts::REGISTRY)
+            let reg_url = self.registries.get(&c.registry).unwrap();
+            if self.service_registry.get(&c.name).is_some() {
+                self.service_registry
+                    .get_mut(&c.name)
                     .unwrap()
-                    .push(u.clone());
+                    .push(reg_url.clone());
             } else {
-                self.services
-                    .insert(consts::REGISTRY.to_string(), vec![u.clone()]);
+                self.service_registry
+                    .insert(c.name.clone(), vec![reg_url.clone()]);
             }
 
             if self.protocols.get(&c.protocol).is_some() {
@@ -126,11 +117,9 @@ impl Dubbo {
         self.init();
 
         // TODO: server registry
-        for (name, url) in self.registries.iter() {
-            tracing::info!("registry name: {:?}, url: {:?}", name, url);
-        }
 
-        let mem_reg = Box::new(RegistryProtocol::new());
+        let mem_reg =
+            Box::new(RegistryProtocol::new().with_services(self.service_registry.clone()));
         let mut async_vec: Vec<Pin<Box<dyn Future<Output = BoxExporter> + Send>>> = Vec::new();
         for (name, items) in self.protocols.iter() {
             for url in items.iter() {

dubbo/src/registry/protocol.rs

@@ -21,7 +21,6 @@ use std::sync::{Arc, RwLock};
 use super::memory_registry::MemoryRegistry;
 use super::BoxRegistry;
 use crate::codegen::TripleInvoker;
-use crate::common::consts;
 use crate::common::url::Url;
 use crate::protocol::triple::triple_exporter::TripleExporter;
 use crate::protocol::triple::triple_protocol::TripleProtocol;
@@ -35,16 +34,24 @@ pub struct RegistryProtocol {
     registries: Arc<RwLock<HashMap<String, BoxRegistry>>>,
     // providerUrl: Exporter
     exporters: Arc<RwLock<HashMap<String, BoxExporter>>>,
+    // serviceName: registryUrls
+    services: HashMap<String, Vec<Url>>,
 }
 
 impl RegistryProtocol {
     pub fn new() -> Self {
         RegistryProtocol {
             registries: Arc::new(RwLock::new(HashMap::new())),
             exporters: Arc::new(RwLock::new(HashMap::new())),
+            services: HashMap::new(),
         }
     }
 
+    pub fn with_services(mut self, services: HashMap<String, Vec<Url>>) -> Self {
+        self.services.extend(services);
+        self
+    }
+
     pub fn get_registry(&mut self, url: Url) -> BoxRegistry {
         let mem = MemoryRegistry::default();
         self.registries
@@ -70,10 +77,14 @@ impl Protocol for RegistryProtocol {
         // init Exporter based on provider_url
         // server registry based on register_url
         // start server health check
-        let registry_url = get_registry_url(url.clone());
-        if !registry_url.protocol.is_empty() {
-            let mut reg = self.get_registry(registry_url.clone());
-            reg.register(registry_url.clone()).unwrap();
+        let registry_url = self.services.get(url.get_service_name().join(",").as_str());
+        if let Some(urls) = registry_url {
+            for url in urls.clone().iter() {
+                if !url.protocol.is_empty() {
+                    let mut reg = self.get_registry(url.clone());
+                    reg.register(url.clone()).unwrap();
+                }
+            }
         }
 
         match url.clone().protocol.as_str() {
@@ -83,10 +94,11 @@ impl Protocol for RegistryProtocol {
             }
             _ => {
                 tracing::error!("protocol {:?} not implemented", url.protocol);
+                Box::new(TripleExporter::new())
             }
         }
-        Box::new(TripleExporter::new())
     }
+
     async fn refer(self, url: Url) -> Self::Invoker {
         // getRegisterUrl
         // get Registry from registry_url
@@ -95,16 +107,3 @@ impl Protocol for RegistryProtocol {
         Box::new(TripleInvoker::new(url))
     }
 }
-
-fn get_registry_url(mut url: Url) -> Url {
-    // registry_url need storage some places
-    if url.protocol == consts::REGISTRY_PROTOCOL {
-        url.protocol = url.get_param("registry".to_string()).unwrap();
-    }
-
-    url
-}
-
-fn get_provider_url(url: Url) -> Url {
-    url
-}