refactor(dubbo): use registry protocol plugin to init protocol

Author: yang20150702

dubbo/src/framework.rs

@@ -20,11 +20,10 @@ use std::pin::Pin;
 
 use futures::future;
 use futures::Future;
-use futures::FutureExt;
 
 use crate::common::url::Url;
-use crate::protocol::triple::triple_protocol::TripleProtocol;
 use crate::protocol::{BoxExporter, Protocol};
+use crate::registry::protocol::RegistryProtocol;
 use dubbo_config::{get_global_config, RootConfig};
 
 // Invoker是否可以基于hyper写一个通用的
@@ -104,20 +103,17 @@ 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 mut async_vec: Vec<Pin<Box<dyn Future<Output = BoxExporter> + Send>>> = Vec::new();
-        for (key, c) in self.protocols.iter() {
-            match key.as_str() {
-                "triple" => {
-                    let pro = Box::new(TripleProtocol::new());
-                    for u in c.iter() {
-                        let tri_fut = pro.clone().export(u.clone()).boxed();
-                        async_vec.push(tri_fut);
-                    }
-                }
-                _ => {
-                    tracing::error!("protocol {:?} not implemented", key);
-                }
+        for (name, items) in self.protocols.iter() {
+            for url in items.iter() {
+                tracing::info!("protocol: {:?}, service url: {:?}", name, url);
+                let exporter = mem_reg.clone().export(url.to_owned());
+                async_vec.push(exporter)
             }
         }
 

dubbo/src/registry/memory_registry.rs

@@ -17,6 +17,7 @@
 
 #![allow(unused_variables, dead_code, missing_docs)]
 use std::collections::HashMap;
+use std::sync::Arc;
 use std::sync::RwLock;
 
 use super::{NotifyListener, Registry};
@@ -27,15 +28,15 @@ use super::{NotifyListener, Registry};
 
 pub const REGISTRY_GROUP_KEY: &str = "registry.group";
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct MemoryRegistry {
-    registries: RwLock<HashMap<String, String>>,
+    registries: Arc<RwLock<HashMap<String, String>>>,
 }
 
 impl MemoryRegistry {
     pub fn new() -> MemoryRegistry {
         MemoryRegistry {
-            registries: RwLock::new(HashMap::new()),
+            registries: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 }

dubbo/src/registry/protocol.rs

@@ -16,39 +16,43 @@
  */
 
 use std::collections::HashMap;
-use std::sync::Arc;
+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;
 use crate::protocol::BoxExporter;
 use crate::protocol::BoxInvoker;
 use crate::protocol::Protocol;
 
 #[derive(Clone, Default)]
 pub struct RegistryProtocol {
     // registerAddr: Registry
-    registries: Arc<HashMap<String, BoxRegistry>>,
+    registries: Arc<RwLock<HashMap<String, BoxRegistry>>>,
     // providerUrl: Exporter
-    exporters: Arc<HashMap<String, BoxExporter>>,
+    exporters: Arc<RwLock<HashMap<String, BoxExporter>>>,
 }
 
 impl RegistryProtocol {
     pub fn new() -> Self {
         RegistryProtocol {
-            registries: Arc::new(HashMap::new()),
-            exporters: Arc::new(HashMap::new()),
+            registries: Arc::new(RwLock::new(HashMap::new())),
+            exporters: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 
-    pub fn get_registry(&self, url: Url) -> BoxRegistry {
-        // self.registries.clone().insert(url.location.clone(), Box::new(MemoryRegistry::default()));
+    pub fn get_registry(&mut self, url: Url) -> BoxRegistry {
+        let mem = MemoryRegistry::default();
+        self.registries
+            .write()
+            .unwrap()
+            .insert(url.location, Box::new(mem.clone()));
 
-        // *(self.registries.get(&url.location).unwrap())
-        Box::new(MemoryRegistry::default())
+        Box::new(mem)
     }
 }
 
@@ -60,12 +64,27 @@ impl Protocol for RegistryProtocol {
         todo!()
     }
 
-    async fn export(self, url: Url) -> BoxExporter {
+    async fn export(mut self, url: Url) -> BoxExporter {
         // getProviderUrl
         // getRegisterUrl
         // 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();
+        }
+
+        match url.clone().protocol.as_str() {
+            "triple" => {
+                let pro = Box::new(TripleProtocol::new());
+                return pro.export(url).await;
+            }
+            _ => {
+                tracing::error!("protocol {:?} not implemented", url.protocol);
+            }
+        }
         Box::new(TripleExporter::new())
     }
     async fn refer(self, url: Url) -> Self::Invoker {