Ftr: exchange layer abstraction design

Cargo.toml

@@ -14,9 +14,8 @@ members = [
   "dubbo-build",
   "remoting/net",
   "remoting/http",
-  "remoting/h2",
   "remoting/zookeeper",
-  "remoting/exchange",
+  "remoting/base",
   "remoting/xds",
   "protocol/dubbo2",
   "protocol/base",
@@ -28,6 +27,8 @@ members = [
 pin-project = "1"
 tokio = "1.0"
 tower = "0.4"
+tower-service = "0.3.1"
+tower-layer = "0.3"
 tokio-stream = "0.1"
 tokio-util = "0.7"
 socket2 = "0.4"
@@ -42,16 +43,20 @@ logger = {path="./common/logger"}
 utils = {path="./common/utils"}
 base = {path="./common/base"}
 remoting-net = {path="./remoting/net"}
-protocol = {path= "protocol/base" }
+remoting-base = {path="./remoting/base"}
+protocol-base = {path= "protocol/base" }
 protocol-dubbo2 = {path="./protocol/dubbo2"}
 protocol-triple = {path="./protocol/triple"}
 registry-zookeeper = {path="./registry/zookeeper"}
 registry-nacos = {path="./registry/nacos"}
 anyhow = "1.0.66"
+thiserror = "1.0.30"
 dubbo = { path = "./dubbo/" }
 bb8 = "0.8.0" # A connecton pool based on tokio
 serde_yaml = "0.9.4" # yaml file parser
 once_cell = "1.16.0"
 itertools = "0.10.1"
+bytes = "1.0"
+
 
 

common/base/src/url.rs

@@ -161,6 +161,10 @@ impl Url {
     pub fn protocol(&self) -> String {
         self.scheme.clone()
     }
+
+    pub fn get_ip_port(&self) -> String {
+        format!("{}:{}", self.ip, self.port)
+    }
 }
 
 impl Display for Url {

dubbo/Cargo.toml

@@ -12,31 +12,30 @@ repository = "https://github.com/apache/dubbo-rust.git"
 [dependencies]
 hyper = { version = "0.14.19", features = ["full"] }
 http = "0.2"
-tower-service = "0.3.1"
+tower-service.workspace = true
 http-body = "0.4.4"
-tower = { version = "0.4.12", features = ["timeout"] }
+tower = { workspace = true, features = ["timeout"] }
 futures-util = "0.3.23"
 futures-core = "0.3.23"
-tokio = { version = "1.0", features = ["rt-multi-thread", "time", "fs", "macros", "net", "signal"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "time", "fs", "macros", "net", "signal"] }
 prost = "0.10.4"
 async-trait = "0.1.56"
-tower-layer = "0.3"
-bytes = "1.0"
+tower-layer.workspace = true
+bytes.workspace = true
 pin-project.workspace = true
 rand = "0.8.5"
 serde_json.workspace = true
 serde = { workspace = true, features = ["derive"] }
-futures = "0.3"
-tracing = "0.1"
-tracing-subscriber = "0.3.15"
+futures.workspace = true
 axum = "0.5.9"
 async-stream = "0.3"
 flate2 = "1.0"
 aws-smithy-http = "0.54.1"
 itertools.workspace = true
 urlencoding.workspace = true
 lazy_static.workspace = true
-base.workspace=true
+base.workspace = true
+logger.workspace = true
 
 dubbo-config = { path = "../config", version = "0.3.0" }
 

dubbo/src/cluster/directory.rs

@@ -27,6 +27,7 @@ use crate::{
     registry::{memory_registry::MemoryNotifyListener, BoxRegistry, RegistryWrapper},
 };
 use base::Url;
+use logger::tracing;
 
 /// Directory.
 ///

dubbo/src/context.rs

@@ -21,6 +21,7 @@ use std::{
     thread,
 };
 
+use logger::tracing;
 use serde_json::Value;
 use state::Container;
 

dubbo/src/filter/context.rs

@@ -17,6 +17,7 @@
 
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use logger::tracing;
 use serde_json::Value;
 
 use crate::{

dubbo/src/filter/timeout.rs

@@ -14,6 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+use logger::tracing;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use crate::{

dubbo/src/framework.rs

@@ -22,10 +22,6 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use base::Url;
-use futures::{future, Future};
-use tracing::{debug, info};
-
 use crate::{
     protocol::{BoxExporter, Protocol},
     registry::{
@@ -34,7 +30,10 @@ use crate::{
         BoxRegistry, Registry,
     },
 };
+use base::Url;
 use dubbo_config::{get_global_config, protocol::ProtocolRetrieve, RootConfig};
+use futures::{future, Future};
+use logger::tracing;
 
 // Invoker是否可以基于hyper写一个通用的
 
@@ -78,10 +77,10 @@ impl Dubbo {
         }
 
         let root_config = self.config.as_ref().unwrap();
-        debug!("global conf: {:?}", root_config);
+        tracing::debug!("global conf: {:?}", root_config);
         // env::set_var("ZOOKEEPER_SERVERS",root_config);
         for (_, service_config) in root_config.provider.services.iter() {
-            info!("init service name: {}", service_config.interface);
+            tracing::info!("init service name: {}", service_config.interface);
             let url = if root_config
                 .protocols
                 .contains_key(service_config.protocol.as_str())
@@ -91,12 +90,12 @@ impl Dubbo {
                     .get_protocol_or_default(service_config.protocol.as_str());
                 let protocol_url =
                     format!("{}/{}", protocol.to_url(), service_config.interface.clone(),);
-                info!("protocol_url: {:?}", protocol_url);
+                tracing::info!("protocol_url: {:?}", protocol_url);
                 Url::from_url(&protocol_url)
             } else {
                 return Err(format!("base {:?} not exists", service_config.protocol).into());
             };
-            info!("url: {:?}", url);
+            tracing::info!("url: {:?}", url);
             if url.is_none() {
                 continue;
             }
@@ -116,7 +115,7 @@ impl Dubbo {
 
     pub async fn start(&mut self) {
         self.init().unwrap();
-        info!("starting...");
+        tracing::info!("starting...");
         // TODO: server registry
         let mem_reg = Box::new(
             RegistryProtocol::new()
@@ -126,7 +125,7 @@ impl Dubbo {
         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() {
-                info!("base: {:?}, service url: {:?}", name, url);
+                tracing::info!("base: {:?}, service url: {:?}", name, url);
                 let exporter = mem_reg.clone().export(url.to_owned());
                 async_vec.push(exporter);
                 //TODO multiple registry

dubbo/src/registry/memory_registry.rs

@@ -17,11 +17,11 @@
 
 #![allow(unused_variables, dead_code, missing_docs)]
 
+use logger::tracing::debug;
 use std::{
     collections::HashMap,
     sync::{Arc, RwLock},
 };
-use tracing::debug;
 
 use base::Url;
 

dubbo/src/registry/protocol.rs

@@ -16,6 +16,7 @@
  */
 
 use base::Url;
+use logger::tracing;
 use std::{
     collections::HashMap,
     fmt::{Debug, Formatter},

dubbo/src/registry/types.rs

@@ -22,7 +22,7 @@ use std::{
 
 use base::Url;
 use itertools::Itertools;
-use tracing::info;
+use logger::tracing::info;
 
 use crate::{
     registry::{BoxRegistry, Registry},

dubbo/src/triple/decode.rs

@@ -20,6 +20,7 @@ use std::{pin::Pin, task::Poll};
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use futures_util::{future, ready, Stream};
 use http_body::Body;
+use logger::tracing;
 
 use super::compression::{decompress, CompressionEncoding};
 use crate::{

dubbo/src/triple/server/builder.rs

@@ -23,6 +23,7 @@ use std::{
 use base::Url;
 use http::{Request, Response, Uri};
 use hyper::body::Body;
+use logger::tracing;
 use tower_service::Service;
 
 use crate::{triple::transport::DubboServer, BoxBody};

dubbo/src/triple/transport/connection.rs

@@ -18,8 +18,8 @@
 use std::task::Poll;
 
 use hyper::client::{conn::Builder, service::Connect};
+use logger::tracing::debug;
 use tower_service::Service;
-use tracing::debug;
 
 use crate::{boxed, triple::transport::connector::get_connector};
 

dubbo/src/triple/transport/connector/http_connector.rs

@@ -22,6 +22,7 @@ use std::{
 
 use http::Uri;
 use hyper::client::connect::dns::Name;
+use logger::tracing;
 use tokio::net::TcpStream;
 use tower_service::Service;
 

dubbo/src/triple/transport/connector/unix_connector.rs

@@ -22,6 +22,7 @@ use std::{
 
 use http::Uri;
 use hyper::client::connect::dns::Name;
+use logger::tracing;
 use tokio::net::UnixStream;
 use tower_service::Service;
 

dubbo/src/triple/transport/listener/mod.rs

@@ -22,6 +22,7 @@ pub mod unix_listener;
 use std::net::SocketAddr;
 
 use async_trait::async_trait;
+use logger::tracing;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use super::io::BoxIO;

dubbo/src/triple/transport/listener/tcp_listener.rs

@@ -21,6 +21,7 @@ use super::Listener;
 use async_trait::async_trait;
 use futures_core::Stream;
 use hyper::server::accept::Accept;
+use logger::tracing;
 use tokio::net::{TcpListener as tokioTcpListener, TcpStream};
 
 pub struct TcpListener {

dubbo/src/triple/transport/listener/unix_listener.rs

@@ -21,6 +21,7 @@ use super::Listener;
 use async_trait::async_trait;
 use futures_core::Stream;
 use hyper::server::accept::Accept;
+use logger::tracing;
 use tokio::net::{UnixListener as tokioUnixListener, UnixStream};
 
 pub struct UnixListener {

dubbo/src/triple/transport/service.rs

@@ -20,6 +20,7 @@ use std::net::SocketAddr;
 use futures_core::Future;
 use http::{Request, Response};
 use hyper::body::Body;
+use logger::tracing;
 use tokio::time::Duration;
 use tower_service::Service;
 

protocol/base/Cargo.toml

@@ -1,10 +1,11 @@
 [package]
-name = "protocol"
+name = "protocol-base"
 version = "0.1.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 dashmap.workspace = true
-base.workspace = true
+base.workspace = true
+thiserror.workspace = true

protocol/base/src/error.rs

@@ -14,24 +14,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+use thiserror::Error;
 
-use std::{
-    error::Error,
-    fmt::{Debug, Display, Formatter},
-};
-
-pub struct InvokerError(String);
-
-impl Debug for InvokerError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.write_str(self.0.as_str())
-    }
+#[derive(Error, Debug)]
+pub enum InvokerError {
+    #[error("unknown invoker error.")]
+    Unknown,
 }
-
-impl Display for InvokerError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.write_str(self.0.as_str())
-    }
-}
-
-impl Error for InvokerError {}

protocol/base/src/lib.rs

@@ -19,3 +19,5 @@ pub mod error;
 pub mod invocation;
 pub mod invoker;
 pub mod output;
+
+pub type ProtocolName = &'static str;

protocol/dubbo2/Cargo.toml

@@ -6,5 +6,5 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-remoting-net.workspace = true
-protocol.workspace = true
+remoting-base.workspace = true
+protocol-base.workspace = true

protocol/triple/Cargo.toml

@@ -7,5 +7,5 @@ edition = "2021"
 
 [dependencies]
 remoting-net.workspace = true
-protocol.workspace = true
+protocol-base.workspace = true
 base.workspace = true

protocol/triple/src/triple_invoker.rs

@@ -15,7 +15,8 @@
  * limitations under the License.
  */
 use base::{Node, Url};
-use protocol::{
+
+use protocol_base::{
     invocation::BoxInvocation,
     invoker::{BaseInvoker, Invoker},
 };

remoting/base/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "remoting-base"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+bytes.workspace = true
+base.workspace = true
+thiserror.workspace = true
+dashmap.workspace = true
+protocol-base.workspace = true
+anyhow.workspace = true