WIP:
diff --git a/Cargo.lock b/Cargo.lock
index f880360..af8e4df 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -266,6 +266,16 @@
]
[[package]]
+name = "gethostname"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1ebd34e35c46e00bb73e81363248d627782724609fe1b6396f553f68fe3862e"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
+[[package]]
name = "getrandom"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -795,9 +805,11 @@
version = "0.1.0"
dependencies = [
"futures",
+ "gethostname",
"prost",
"prost-types",
"rustls",
+ "thiserror",
"tokio",
"tokio-rustls",
"tonic",
@@ -970,6 +982,26 @@
]
[[package]]
+name = "thiserror"
+version = "1.0.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "854babe52e4df1653706b98fcfc05843010039b406875930a70e4d9644e5c417"
+dependencies = [
+ "thiserror-impl",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa32fd3f627f367fe16f893e2597ae3c05020f8bba2666a4e6ea73d377e5714b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
name = "tokio"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index 52cdba9..1a31fac 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,6 +13,8 @@
tokio = { version = "1", features = ["full"]}
tokio-rustls = "0.23"
rustls = {version = "0.20", features = ["default", "dangerous_configuration"]}
+gethostname = "0.2"
+thiserror = "1.0"
[build-dependencies]
tonic-build = {version = "0.7", features = ["default", "compression"]}
@@ -20,4 +22,8 @@
[[bin]]
name = "server"
-path = "src/bin/server.rs"
\ No newline at end of file
+path = "src/bin/server.rs"
+
+[[bin]]
+name = "task"
+path = "src/bin/task.rs"
\ No newline at end of file
diff --git a/src/bin/task.rs b/src/bin/task.rs
new file mode 100644
index 0000000..bf23764
--- /dev/null
+++ b/src/bin/task.rs
@@ -0,0 +1,25 @@
+use std::sync::{
+ atomic::{self, Ordering},
+ Arc,
+};
+use tokio::time;
+
+#[tokio::main]
+async fn main() {
+ let stopped = Arc::new(atomic::AtomicBool::new(false));
+
+ let stop_flag = Arc::clone(&stopped);
+ let handle = tokio::spawn(async move {
+ while (!stop_flag.load(Ordering::Relaxed)) {
+ tokio::time::sleep(time::Duration::from_secs(1)).await;
+ println!("Ping");
+ }
+ });
+
+ let terminate_task = tokio::spawn(async move {
+ tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
+ stopped.store(true, Ordering::Relaxed);
+ });
+
+ terminate_task.await;
+}
diff --git a/src/client.rs b/src/client.rs
index 3b1cdde..0b0bd85 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -2,6 +2,7 @@
messaging_service_client::MessagingServiceClient, QueryRouteRequest, QueryRouteResponse,
SendMessageRequest, SendMessageResponse,
};
+use rustls::client;
use tonic::{
metadata::MetadataMap,
transport::{Channel, ClientTlsConfig},
@@ -9,38 +10,93 @@
};
use crate::credentials::CredentialProvider;
+use std::collections::HashMap;
+use std::rc::Rc;
+use std::{
+ sync::{
+ atomic::{AtomicUsize, Ordering},
+ Arc, Mutex, RwLock,
+ },
+ thread,
+};
-#[derive(Default)]
-struct ClientConfig {
+static CLIENT_SEQUENCE: AtomicUsize = AtomicUsize::new(0);
+
+pub struct ClientConfig {
region: String,
service_name: String,
- resource_namespace: String,
+ resource_namespace: Option<String>,
credential_provider: Option<Box<dyn CredentialProvider>>,
- tenant_id: String,
+ tenant_id: Option<String>,
+ connect_timeout: std::time::Duration,
io_timeout: std::time::Duration,
long_polling_timeout: std::time::Duration,
- group: String,
+ group: Option<String>,
client_id: String,
tracing: bool,
}
+fn build_client_id() -> String {
+ let mut client_id = String::new();
+ match gethostname::gethostname().into_string() {
+ Ok(hostname) => {
+ client_id.push_str(&hostname);
+ }
+ Err(_) => {
+ client_id.push_str("localhost");
+ }
+ };
+ client_id.push('@');
+ let pid = std::process::id();
+ client_id.push_str(&pid.to_string());
+ client_id.push('#');
+ let sequence = CLIENT_SEQUENCE.fetch_add(1usize, Ordering::Relaxed);
+ client_id.push_str(&sequence.to_string());
+ client_id
+}
+
+impl Default for ClientConfig {
+ fn default() -> Self {
+ let client_id = build_client_id();
+ Self {
+ region: String::from("cn-hangzhou"),
+ service_name: String::from("RocketMQ"),
+ resource_namespace: None,
+ credential_provider: None,
+ tenant_id: None,
+ connect_timeout: std::time::Duration::from_secs(3),
+ io_timeout: std::time::Duration::from_secs(3),
+ long_polling_timeout: std::time::Duration::from_secs(3),
+ group: None,
+ client_id,
+ tracing: false,
+ }
+ }
+}
+
pub struct RpcClient {
+ client_config: Arc<RwLock<ClientConfig>>,
stub: MessagingServiceClient<Channel>,
peer_address: String,
// client_config: std::rc::Rc<ClientConfig>,
}
impl RpcClient {
- pub async fn new(target: String) -> Result<RpcClient, Box<dyn std::error::Error>> {
+ pub async fn new(
+ target: String,
+ client_config: Arc<RwLock<ClientConfig>>,
+ ) -> Result<RpcClient, Box<dyn std::error::Error>> {
+ let config = Arc::clone(&client_config);
let mut channel = Channel::from_shared(target.clone())?
.tcp_nodelay(true)
- .connect_timeout(std::time::Duration::from_secs(3));
+ .connect_timeout(config.read().unwrap().connect_timeout);
if target.starts_with("https://") {
channel = channel.tls_config(ClientTlsConfig::new())?;
}
let channel = channel.connect().await?;
let stub = MessagingServiceClient::new(channel);
Ok(RpcClient {
+ client_config,
stub,
peer_address: target,
})
@@ -61,13 +117,16 @@
&mut self,
request: SendMessageRequest,
) -> Result<Response<SendMessageResponse>, Box<dyn std::error::Error>> {
- let mut req = Request::new(request);
+ let req = Request::new(request);
Ok(self.stub.send_message(req).await?)
}
}
#[derive(Default)]
-pub struct ClientManager {}
+pub struct ClientManager {
+ client_config: Arc<RwLock<ClientConfig>>,
+ clients: Mutex<HashMap<String, Rc<RpcClient>>>,
+}
impl ClientManager {
pub async fn start(&self) {
@@ -80,25 +139,48 @@
});
let _result = handle.await;
}
+
+ pub async fn get_rpc_client(
+ &'static mut self,
+ endpoint: &str,
+ ) -> Result<Rc<RpcClient>, Box<dyn std::error::Error>> {
+ let mut rpc_clients = self.clients.lock()?;
+ let key = endpoint.to_owned();
+ match rpc_clients.get(&key) {
+ Some(value) => {
+ return Ok(Rc::clone(value));
+ }
+ None => {
+ let rpc_client =
+ RpcClient::new(key.clone(), Arc::clone(&self.client_config)).await?;
+ let client = Rc::new(rpc_client);
+ rpc_clients.insert(key, Rc::clone(&client));
+ Ok(client)
+ }
+ }
+ }
}
#[cfg(test)]
mod test {
use super::*;
+ use std::collections::HashSet;
use crate::pb::{Code, Resource};
#[tokio::test]
async fn test_connect() {
let target = "http://127.0.0.1:5001";
- let _rpc_client = RpcClient::new(target.to_owned())
+ let client_config = Arc::new(RwLock::new(ClientConfig::default()));
+ let _rpc_client = RpcClient::new(target.to_owned(), client_config)
.await
.expect("Should be able to connect");
}
#[tokio::test]
async fn test_connect_staging() {
+ let client_config = Arc::new(RwLock::new(ClientConfig::default()));
let target = "https://mq-inst-1080056302921134-bxuibml7.mq.cn-hangzhou.aliyuncs.com:80";
- let _rpc_client = RpcClient::new(target.to_owned())
+ let _rpc_client = RpcClient::new(target.to_owned(), client_config)
.await
.expect("Failed to connect to staging proxy server");
}
@@ -106,7 +188,8 @@
#[tokio::test]
async fn test_query_route() {
let target = "http://127.0.0.1:5001";
- let mut rpc_client = RpcClient::new(target.to_owned())
+ let client_config = Arc::new(RwLock::new(ClientConfig::default()));
+ let mut rpc_client = RpcClient::new(target.to_owned(), client_config)
.await
.expect("Should be able to connect");
let topic = Resource {
@@ -126,6 +209,17 @@
assert_eq!(route_response.status.unwrap().code, Code::Ok as i32);
}
+ #[test]
+ fn test_build_client_id() {
+ let mut set = HashSet::new();
+ let cnt = 1000;
+ for _ in 0..cnt {
+ let client_id = build_client_id();
+ set.insert(client_id);
+ }
+ assert_eq!(cnt, set.len());
+ }
+
#[tokio::test]
async fn test_periodic_task() {
let client_manager = ClientManager::default();
diff --git a/src/credentials.rs b/src/credentials.rs
index 43d44a9..27812f8 100644
--- a/src/credentials.rs
+++ b/src/credentials.rs
@@ -82,6 +82,5 @@
fn test_environment_variable_credential_provider() {
let env_credentials_provider = EnvironmentVariableCredentialProvider::new();
assert_eq!(true, env_credentials_provider.is_err());
-
}
}
diff --git a/src/error.rs b/src/error.rs
new file mode 100644
index 0000000..6d4fb44
--- /dev/null
+++ b/src/error.rs
@@ -0,0 +1,7 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ClientError {
+ #[error("Host name inconvertible to UTF-8")]
+ InvalidHostName,
+}
diff --git a/src/lib.rs b/src/lib.rs
index 7ffd5e9..a7427ff 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,5 @@
pub mod client;
pub mod credentials;
+pub mod error;
pub mod pb;
pub mod server;
