Rft: add Cluster to ClientBuilder

dubbo-build/src/client.rs

@@ -90,11 +90,6 @@ pub fn generate<T: Service>(
  }
  }
 
- pub fn with_cluster(mut self, invoker: ClusterInvoker) -> Self {
- self.inner = self.inner.with_cluster(invoker);
- self
- }
-
  #methods
 
  }

dubbo/Cargo.toml

@@ -10,7 +10,7 @@ repository = "https:/apache/dubbo-rust.git"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-hyper = { version = "0.14.19", features = ["full"] }
+hyper = { version = "0.14.26", features = ["full"] }
 http = "0.2"
 tower-service.workspace = true
 http-body = "0.4.4"
@@ -33,7 +33,7 @@ futures.workspace = true
 axum = "0.5.9"
 async-stream = "0.3"
 flate2 = "1.0"
-aws-smithy-http = "0.54.1"
+aws-smithy-http = "0.55.2"
 dyn-clone = "1.0.11"
 itertools.workspace = true
 urlencoding.workspace = true

dubbo/src/cluster/directory.rs

@@ -23,39 +23,21 @@ use std::{
 };
 
 use crate::{
+ codegen::TripleInvoker,
  invocation::{Invocation, RpcInvocation},
- registry::{memory_registry::MemoryNotifyListener, BoxRegistry, RegistryWrapper},
+ protocol::BoxInvoker,
+ registry::{memory_registry::MemoryNotifyListener, BoxRegistry},
 };
 use dubbo_base::Url;
 use dubbo_logger::tracing;
 
+use crate::cluster::Directory;
+
 /// Directory.
 ///
 /// [Directory Service](http://en.wikipedia.org/wiki/Directory_service)
-pub trait Directory: Debug + DirectoryClone {
- fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<Url>;
-}
 
-pub trait DirectoryClone {
- fn clone_box(&self) -> Box<dyn Directory>;
-}
-
-impl<T> DirectoryClone for T
-where
- T: 'static + Directory + Clone,
-{
- fn clone_box(&self) -> Box<dyn Directory> {
- Box::new(self.clone())
- }
-}
-
-impl Clone for Box<dyn Directory> {
- fn clone(&self) -> Box<dyn Directory> {
- self.clone_box()
- }
-}
-
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct StaticDirectory {
  uri: http::Uri,
 }
@@ -78,51 +60,36 @@ impl StaticDirectory {
 }
 
 impl Directory for StaticDirectory {
- fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<Url> {
+ fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<BoxInvoker> {
  let url = Url::from_url(&format!(
  "tri://{}:{}/{}",
  self.uri.host().unwrap(),
  self.uri.port().unwrap(),
  invocation.get_target_service_unique_name(),
  ))
  .unwrap();
- vec![url]
- }
-}
-
-impl DirectoryClone for StaticDirectory {
- fn clone_box(&self) -> Box<dyn Directory> {
- Box::new(StaticDirectory {
- uri: self.uri.clone(),
- })
+ let invoker = Box::new(TripleInvoker::new(url));
+ vec![invoker]
  }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct RegistryDirectory {
- registry: RegistryWrapper,
+ registry: Arc<BoxRegistry>,
  service_instances: Arc<RwLock<HashMap<String, Vec<Url>>>>,
 }
 
 impl RegistryDirectory {
  pub fn new(registry: BoxRegistry) -> RegistryDirectory {
  RegistryDirectory {
- registry: RegistryWrapper {
- registry: Some(registry),
- },
+ registry: Arc::new(registry),
  service_instances: Arc::new(RwLock::new(HashMap::new())),
  }
  }
 }
 
-impl DirectoryClone for RegistryDirectory {
- fn clone_box(&self) -> Box<dyn Directory> {
- todo!()
- }
-}
-
 impl Directory for RegistryDirectory {
- fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<Url> {
+ fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<BoxInvoker> {
  let service_name = invocation.get_target_service_unique_name();
 
  let url = Url::from_url(&format!(
@@ -132,9 +99,6 @@ impl Directory for RegistryDirectory {
  .unwrap();
 
  self.registry
- .registry
- .as_ref()
- .expect("msg")
  .subscribe(
  url,
  Arc::new(MemoryNotifyListener {
@@ -149,6 +113,11 @@ impl Directory for RegistryDirectory {
  .expect("service_instances.read");
  let binding = Vec::new();
  let url_vec = map.get(&service_name).unwrap_or(&binding);
- url_vec.to_vec()
+ // url_vec.to_vec()
+ let mut invokers: Vec<BoxInvoker> = vec![];
+ for item in url_vec.iter() {
+ invokers.push(Box::new(TripleInvoker::new(item.clone())));
+ }
+ invokers
  }
 }

dubbo/src/cluster/mod.rs

@@ -19,6 +19,7 @@ use std::{collections::HashMap, fmt::Debug, sync::Arc, task::Poll};
 
 use aws_smithy_http::body::SdkBody;
 use dubbo_base::Url;
+use dyn_clone::DynClone;
 
 use crate::{
  empty_body,
@@ -28,13 +29,14 @@ use crate::{
 
 pub mod directory;
 pub mod loadbalance;
-pub mod support;
 
-pub trait Directory: Debug {
+pub trait Directory: Debug + DynClone {
  fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<BoxInvoker>;
- fn is_empty(&self) -> bool;
+ // fn is_empty(&self) -> bool;
 }
 
+dyn_clone::clone_trait_object!(Directory);
+
 type BoxDirectory = Box<dyn Directory + Send + Sync>;
 
 pub trait Cluster {
@@ -76,21 +78,20 @@ impl Invoker<http::Request<SdkBody>> for FailoverCluster {
  }
 
  fn call(&mut self, req: http::Request<SdkBody>) -> Self::Future {
- println!("req: {}", req.body().content_length().unwrap());
- let clone_body = req.body().try_clone().unwrap();
- let mut clone_req = http::Request::builder()
- .uri(req.uri().clone())
- .method(req.method().clone());
- *clone_req.headers_mut().unwrap() = req.headers().clone();
- let r = clone_req.body(clone_body).unwrap();
+ // let clone_body = req.body().try_clone().unwrap();
+ // let mut clone_req = http::Request::builder()
+ // .uri(req.uri().clone())
+ // .method(req.method().clone());
+ // *clone_req.headers_mut().unwrap() = req.headers().clone();
+ // let r = clone_req.body(clone_body).unwrap();
  let invokers = self.dir.list(
  RpcInvocation::default()
  .with_service_unique_name("hello".to_string())
  .into(),
  );
  for mut invoker in invokers {
  let fut = async move {
- let res = invoker.call(r).await;
+ let res = invoker.call(req).await;
  return res;
  };
  return Box::pin(fut);
@@ -110,7 +111,7 @@ impl Invoker<http::Request<SdkBody>> for FailoverCluster {
  }
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct MockDirectory {
  // router_chain: RouterChain,
  invokers: Vec<BoxInvoker>,
@@ -134,9 +135,9 @@ impl Directory for MockDirectory {
  self.invokers.clone()
  }
 
- fn is_empty(&self) -> bool {
- false
- }
+ // fn is_empty(&self) -> bool {
+ //  false
+ // }
 }
 
 #[derive(Debug, Default)]