Rft(dubbo): impl memory_protocol, update registry config

Author: yang20150702

config/src/config.rs

@@ -37,6 +37,7 @@ pub struct RootConfig {
     pub name: String,
     pub service: HashMap<String, ServiceConfig>,
     pub protocols: HashMap<String, ProtocolConfig>,
+    pub registries: HashMap<String, String>,
 
     #[serde(skip_serializing, skip_deserializing)]
     pub data: HashMap<String, String>,
@@ -63,6 +64,7 @@ impl RootConfig {
             name: "dubbo".to_string(),
             service: HashMap::new(),
             protocols: HashMap::new(),
+            registries: HashMap::new(),
             data: HashMap::new(),
         }
     }

dubbo/src/common/consts.rs

@@ -0,0 +1,18 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+
+pub const REGISTRY_PROTOCOL: &str = "registry";

dubbo/src/common/mod.rs

@@ -15,4 +15,5 @@
  * limitations under the License.
  */
 
+pub mod consts;
 pub mod url;

dubbo/src/framework.rs

@@ -32,6 +32,7 @@ use dubbo_config::{get_global_config, RootConfig};
 #[derive(Default)]
 pub struct Dubbo {
     protocols: HashMap<String, Vec<Url>>,
+    registries: HashMap<String, Url>,
     config: Option<RootConfig>,
 }
 
@@ -40,6 +41,7 @@ impl Dubbo {
         tracing_subscriber::fmt::init();
         Self {
             protocols: HashMap::new(),
+            registries: HashMap::new(),
             config: None,
         }
     }
@@ -56,6 +58,12 @@ impl Dubbo {
 
         let conf = self.config.as_ref().unwrap();
         tracing::debug!("global conf: {:?}", conf);
+
+        for (name, url) in conf.registries.iter() {
+            self.registries
+                .insert(name.to_string(), Url::from_url(url).unwrap());
+        }
+
         for (_, c) in conf.service.iter() {
             let u = if c.protocol_configs.is_empty() {
                 let protocol = match conf.protocols.get(&c.protocol) {

dubbo/src/protocol/mod.rs

@@ -51,7 +51,16 @@ pub trait Invoker<ReqBody> {
     fn call(&mut self, req: ReqBody) -> Self::Future;
 }
 
-pub type BoxExporter = Box<dyn Exporter>;
+pub type BoxExporter = Box<dyn Exporter + Send + Sync>;
+pub type BoxInvoker = Box<
+    dyn Invoker<
+            http::Request<hyper::Body>,
+            Response = http::Response<crate::BoxBody>,
+            Error = crate::Error,
+            Future = crate::BoxFuture<http::Response<crate::BoxBody>, crate::Error>,
+        > + Send
+        + Sync,
+>;
 
 pub struct WrapperInvoker<T>(T);
 

dubbo/src/registry/mod.rs

@@ -42,4 +42,5 @@ pub struct ServiceEvent {
     service: Url,
 }
 
-pub type BoxRegistry = Box<dyn Registry<NotifyListener = memory_registry::MemoryNotifyListener>>;
+pub type BoxRegistry =
+    Box<dyn Registry<NotifyListener = memory_registry::MemoryNotifyListener> + Send + Sync>;

dubbo/src/registry/protocol.rs

@@ -16,19 +16,75 @@
  */
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
+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::BoxExporter;
+use crate::protocol::BoxInvoker;
+use crate::protocol::Protocol;
 
+#[derive(Clone, Default)]
 pub struct RegistryProtocol {
-    registries: HashMap<String, BoxRegistry>,
+    // registerAddr: Registry
+    registries: Arc<HashMap<String, BoxRegistry>>,
+    // providerUrl: Exporter
+    exporters: Arc<HashMap<String, BoxExporter>>,
 }
 
 impl RegistryProtocol {
+    pub fn new() -> Self {
+        RegistryProtocol {
+            registries: Arc::new(HashMap::new()),
+            exporters: Arc::new(HashMap::new()),
+        }
+    }
+
+    pub fn get_registry(&self, url: Url) -> BoxRegistry {
+        // self.registries.clone().insert(url.location.clone(), Box::new(MemoryRegistry::default()));
+
+        // *(self.registries.get(&url.location).unwrap())
+        Box::new(MemoryRegistry::default())
+    }
+}
+
+#[async_trait::async_trait]
+impl Protocol for RegistryProtocol {
+    type Invoker = BoxInvoker;
+
     fn destroy(&self) {
         todo!()
     }
 
-    async fn export(self, url: Url) {}
-    async fn refer(self, url: Url) {}
+    async fn export(self, url: Url) -> BoxExporter {
+        // getProviderUrl
+        // getRegisterUrl
+        // init Exporter based on provider_url
+        // server registry based on register_url
+        // start server health check
+        Box::new(TripleExporter::new())
+    }
+    async fn refer(self, url: Url) -> Self::Invoker {
+        // getRegisterUrl
+        // get Registry from registry_url
+        // init directory based on registry_url and Registry
+        // init Cluster based on Directory generates Invoker
+        Box::new(TripleInvoker::new(url))
+    }
+}
+
+fn get_registry_url(mut url: Url) -> Url {
+    if url.protocol == consts::REGISTRY_PROTOCOL {
+        url.protocol = url.get_param("registry".to_string()).unwrap();
+    }
+
+    url
+}
+
+fn get_provider_url(url: Url) -> Url {
+    url
 }