merge main, fix conflicts, del deprecated

.gitignore

@@ -13,8 +13,8 @@ Cargo.lock
 .idea
 .vscode
 .history
+.DS_Store
 
 # Customize
 ### google.protobuf.rs build by prost-build, exclude it because no content.
 **/google.protobuf.rs
-.DS_Store

CHANGELOG.md

@@ -1,5 +1,25 @@
 # 变更日志 | Change log
 
+### 0.3.0
+- Refactor: tonic instead of tikv/grpc-rs
+- TODO
+
+### 0.2.6
+
+- 修复 `ServiceInfoUpdateTask` 丢失 auth header
+
+---
+
+- fix lose auth headers in ServiceInfoUpdateTask
+
+### 0.2.5
+
+- 优化重连机制
+
+---
+
+- Enhance: optimize reconnect logic
+
 ### 0.2.4
 
 - 清理无用代码

Cargo.toml

@@ -18,7 +18,7 @@
 
 [package]
 name = "nacos-sdk"
-version = "0.2.4"
+version = "0.3.0-alpha"
 edition = "2021"
 authors = ["nacos-group", "CheirshCai <[email protected]>", "onewe <[email protected]>"]
 license = "Apache-2.0"

README.md

@@ -18,7 +18,7 @@ Add the dependency in `Cargo.toml`:
 ```toml
 [dependencies]
 # If you need async API, which can be enabled via `features = ["async"]`
-nacos-sdk = { version = "0.2", features = ["default"] }
+nacos-sdk = { version = "0.3", features = ["default"] }
 ```
 
 ### Usage of Config

examples/simple_app.rs

@@ -6,8 +6,6 @@ use nacos_sdk::api::naming::{
     NamingChangeEvent, NamingEventListener, NamingService, NamingServiceBuilder, ServiceInstance,
 };
 use nacos_sdk::api::props::ClientProps;
-use std::time::Duration;
-use tokio::time::sleep;
 
 /// enable https auth run with command:
 /// cargo run --example simple_app --features default,tls
@@ -73,7 +71,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Some(constants::DEFAULT_GROUP.to_string()),
         vec![service_instance1],
     );
-    sleep(Duration::from_millis(111)).await;
+    tokio::time::sleep(tokio::time::Duration::from_millis(666)).await;
 
     let instances_ret = naming_service.get_all_instances(
         "test-service".to_string(),
@@ -86,7 +84,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Err(err) => tracing::error!("naming get_all_instances error {:?}", err),
     }
 
-    sleep(Duration::from_secs(300)).await;
+    tokio::time::sleep(tokio::time::Duration::from_secs(300)).await;
 
     Ok(())
 }

src/api/naming.rs

@@ -145,13 +145,6 @@ pub trait NamingEventListener: Send + Sync + 'static {
 #[doc(alias("naming", "sdk", "api"))]
 #[cfg(not(feature = "async"))]
 pub trait NamingService {
-    #[deprecated(since = "0.2.2", note = "Users should instead use register_instance")]
-    fn register_service(
-        &self,
-        service_name: String,
-        group_name: Option<String>,
-        service_instance: ServiceInstance,
-    ) -> Result<()>;
 
     fn register_instance(
         &self,
@@ -182,16 +175,6 @@ pub trait NamingService {
         subscribe: bool,
     ) -> Result<Vec<ServiceInstance>>;
 
-    #[deprecated(since = "0.2.2", note = "Users should instead use select_instances")]
-    fn select_instance(
-        &self,
-        service_name: String,
-        group_name: Option<String>,
-        clusters: Vec<String>,
-        subscribe: bool,
-        healthy: bool,
-    ) -> Result<Vec<ServiceInstance>>;
-
     fn select_instances(
         &self,
         service_name: String,

src/api/plugin/auth/auth_by_http.rs

@@ -66,6 +66,7 @@ impl AuthPlugin for HttpLoginAuthPlugin {
 
         tracing::debug!("Http login with username={username},password={password}");
 
+        let (sender, receiver) = std::sync::mpsc::channel::<Option<HttpLoginResponse>>();
         let future = async move {
             let resp = reqwest::Client::new()
                 .post(login_url)
@@ -75,26 +76,27 @@ impl AuthPlugin for HttpLoginAuthPlugin {
             tracing::debug!("Http login resp={resp:?}");
 
             if resp.is_err() {
-                return None;
+                sender.send(None).expect("send response failed");
+                return;
             }
 
             let resp_text = resp.unwrap().text().await.unwrap();
 
             let resp_obj = serde_json::from_str::<HttpLoginResponse>(&resp_text);
             if resp_obj.is_err() {
-                return None;
+                tracing::error!("Http login error with resp_text={resp_text}");
+                sender.send(None).expect("send response failed");
+                return;
             }
-            Some(resp_obj.unwrap())
+            sender
+                .send(Some(resp_obj.unwrap()))
+                .expect("send response failed");
         };
 
-        let login_response = futures::executor::block_on(crate::common::executor::spawn(future));
+        crate::common::executor::spawn(future);
+        let login_response = receiver.recv().expect("receive response failed");
 
-        if let Err(e) = login_response.as_ref() {
-            tracing::error!("Spawn Http login task failed. {e:?}");
-            return;
-        }
-
-        if let Ok(Some(login_response)) = login_response {
+        if let Some(login_response) = login_response {
             let delay_sec = login_response.token_ttl / 10;
             let new_login_identity = LoginIdentityContext::default()
                 .add_context(ACCESS_TOKEN, login_response.access_token);
@@ -136,7 +138,7 @@ mod tests {
             .init();
 
         let http_auth_plugin = HttpLoginAuthPlugin::default();
-        http_auth_plugin.set_server_list(vec!["127.0.0.1:8848".to_string()]);
+        http_auth_plugin.set_server_list(vec!["0.0.0.0:8848".to_string()]);
 
         let auth_context = AuthContext::default()
             .add_param(crate::api::plugin::USERNAME, "nacos")

src/common/remote/grpc/tonic.rs

@@ -30,8 +30,8 @@ use super::{
     server_address::ServerAddress,
 };
 use crate::api::error::Error;
-use crate::api::error::Error::TonicGrpcStatus;
 use crate::api::error::Error::NoAvailableServer;
+use crate::api::error::Error::TonicGrpcStatus;
 
 #[derive(Clone)]
 pub(crate) struct Tonic {
@@ -300,12 +300,15 @@ impl Service<NacosGrpcCall> for Tonic {
 
     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if !self.server_address.is_available() {
-            error!("the server address {}:{} is not available", self.server_address.host(), self.server_address.port());
+            error!(
+                "the server address {}:{} is not available",
+                self.server_address.host(),
+                self.server_address.port()
+            );
             Poll::Ready(Err(NoAvailableServer))
         } else {
             Poll::Ready(Ok(()))
         }
-
     }
 
     fn call(&mut self, call: NacosGrpcCall) -> Self::Future {

src/lib.rs

@@ -28,7 +28,7 @@
 //! - If you need async API, which can be enabled via `features = ["async"]`
 //! ```toml
 //! [dependencies]
-//! nacos-sdk = { version = "0.2", features = ["default"] }
+//! nacos-sdk = { version = "0.3", features = ["default"] }
 //! ```
 //!
 //! ## General Configurations and Initialization

src/naming/mod.rs

@@ -572,16 +572,6 @@ impl NacosNamingService {
 
 #[cfg(not(feature = "async"))]
 impl NamingService for NacosNamingService {
-    #[instrument(fields(client_id = &self.client_id, group = group_name, service_name = service_name), skip_all)]
-    fn register_service(
-        &self,
-        service_name: String,
-        group_name: Option<String>,
-        service_instance: ServiceInstance,
-    ) -> Result<()> {
-        let future = self.register_instance_async(service_name, group_name, service_instance);
-        futures::executor::block_on(future)
-    }
 
     #[instrument(fields(client_id = &self.client_id, group = group_name, service_name = service_name), skip_all)]
     fn deregister_instance(
@@ -618,20 +608,6 @@ impl NamingService for NacosNamingService {
         futures::executor::block_on(future)
     }
 
-    #[instrument(fields(client_id = &self.client_id, group = group_name, service_name = service_name), skip_all)]
-    fn select_instance(
-        &self,
-        service_name: String,
-        group_name: Option<String>,
-        clusters: Vec<String>,
-        subscribe: bool,
-        healthy: bool,
-    ) -> Result<Vec<ServiceInstance>> {
-        let future =
-            self.select_instances_async(service_name, group_name, clusters, subscribe, healthy);
-        futures::executor::block_on(future)
-    }
-
     #[instrument(fields(client_id = &self.client_id, group = group_name, service_name = service_name), skip_all)]
     fn select_one_healthy_instance(
         &self,