feat: unsubscribe service

src/api/events/mod.rs

@@ -1,8 +1,5 @@
-use std::hash::Hash;
 use std::{
     any::{Any, TypeId},
-    collections::hash_map::DefaultHasher,
-    hash::Hasher,
     sync::Arc,
 };
 pub mod naming;
@@ -21,29 +18,9 @@ pub trait Subscriber: Send + Sync + 'static {
     fn event_type(&self) -> TypeId;
 
     fn subscriber_type(&self) -> TypeId;
-
-    fn as_any(&self) -> &dyn Any;
-
-    fn eq_subscriber(&self, other: &dyn Subscriber) -> bool;
-
-    fn hash_value(&self) -> u64;
 }
 
-impl PartialEq for dyn Subscriber {
-    fn eq(&self, other: &Self) -> bool {
-        self.eq_subscriber(other)
-    }
-}
-
-impl Hash for dyn Subscriber {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.hash_value().hash(state);
-    }
-}
-
-impl Eq for dyn Subscriber {}
-
-impl<T: NacosEventSubscriber + PartialEq + Hash> Subscriber for T {
+impl<T: NacosEventSubscriber> Subscriber for T {
     fn on_event(&self, event: Arc<Box<dyn NacosEvent>>) {
         let event = event.as_any().downcast_ref::<T::EventType>();
         if event.is_none() {
@@ -60,23 +37,6 @@ impl<T: NacosEventSubscriber + PartialEq + Hash> Subscriber for T {
     fn subscriber_type(&self) -> TypeId {
         TypeId::of::<T>()
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn eq_subscriber(&self, other: &dyn Subscriber) -> bool {
-        other
-            .as_any()
-            .downcast_ref::<T>()
-            .map_or(false, |subscriber| self == subscriber)
-    }
-
-    fn hash_value(&self) -> u64 {
-        let mut hasher = DefaultHasher::new();
-        self.hash(&mut hasher);
-        hasher.finish()
-    }
 }
 
 pub trait NacosEventSubscriber: Send + Sync + 'static {

src/api/naming.rs

@@ -217,6 +217,22 @@ pub trait NamingService {
         clusters: Vec<String>,
         subscriber: Arc<Box<dyn Subscriber>>,
     ) -> AsyncFuture<()>;
+
+    fn unsubscribe(
+        &self,
+        service_name: String,
+        group_name: Option<String>,
+        clusters: Vec<String>,
+        subscriber: Arc<Box<dyn Subscriber>>,
+    ) -> Result<()>;
+
+    fn unsubscribe_async(
+        &self,
+        service_name: String,
+        group_name: Option<String>,
+        clusters: Vec<String>,
+        subscriber: Arc<Box<dyn Subscriber>>,
+    ) -> AsyncFuture<()>;
 }
 
 pub struct NamingServiceBuilder {

src/common/event_bus/mod.rs

@@ -7,7 +7,7 @@ pub(self) mod __private {
     use lazy_static::lazy_static;
     use std::{
         any::TypeId,
-        collections::{HashMap, HashSet},
+        collections::HashMap,
         sync::{Arc, RwLock},
     };
     use tokio::sync::mpsc::{channel, Receiver, Sender};
@@ -22,7 +22,7 @@ pub(self) mod __private {
         pub static ref EVENT_BUS: EventBus = EventBus::new();
     }
 
-    type SubscribersContainerType = Arc<RwLock<HashMap<TypeId, HashSet<Arc<Box<dyn Subscriber>>>>>>;
+    type SubscribersContainerType = Arc<RwLock<HashMap<TypeId, Vec<Arc<Box<dyn Subscriber>>>>>>;
 
     pub struct EventBus {
         subscribers: SubscribersContainerType,
@@ -33,10 +33,9 @@ pub(self) mod __private {
         pub fn new() -> Self {
             let (sender, receiver) = channel::<Box<dyn NacosEvent>>(2048);
 
-            let subscribers = Arc::new(RwLock::new(HashMap::<
-                TypeId,
-                HashSet<Arc<Box<dyn Subscriber>>>,
-            >::new()));
+            let subscribers = Arc::new(RwLock::new(
+                HashMap::<TypeId, Vec<Arc<Box<dyn Subscriber>>>>::new(),
+            ));
             Self::hand_event(receiver, subscribers.clone());
             EventBus {
                 subscribers: subscribers.clone(),
@@ -93,13 +92,12 @@ pub(self) mod __private {
 
             let key = subscriber.event_type();
 
-            let set = lock_guard.get_mut(&key);
-            if let Some(set) = set {
-                set.insert(subscriber);
+            let vec = lock_guard.get_mut(&key);
+            if let Some(vec) = vec {
+                vec.push(subscriber);
             } else {
-                let mut set = HashSet::default();
-                set.insert(subscriber);
-                lock_guard.insert(key, set);
+                let vec = vec![subscriber];
+                lock_guard.insert(key, vec);
             }
         }
 
@@ -113,14 +111,31 @@ pub(self) mod __private {
 
             let key = subscriber.event_type();
 
-            let set = lock_guard.get_mut(&key);
+            let vec = lock_guard.get_mut(&key);
 
-            if set.is_none() {
+            if vec.is_none() {
                 return;
             }
 
-            let set = set.unwrap();
-            set.remove(&subscriber);
+            let vec = vec.unwrap();
+
+            let index = self.index_of_subscriber(vec, &subscriber);
+            if let Some(index) = index {
+                vec.remove(index);
+            }
+        }
+
+        fn index_of_subscriber(
+            &self,
+            vec: &[Arc<Box<dyn Subscriber>>],
+            target: &Arc<Box<dyn Subscriber>>,
+        ) -> Option<usize> {
+            for (index, subscriber) in vec.iter().enumerate() {
+                if Arc::ptr_eq(subscriber, target) {
+                    return Some(index);
+                }
+            }
+            None
         }
     }
 }
@@ -182,4 +197,30 @@ mod tests {
         let three_millis = time::Duration::from_secs(3);
         thread::sleep(three_millis);
     }
+
+    #[test]
+    pub fn test_register_and_unregister() {
+        let event = NamingChangeEvent {
+            message: "register".to_owned(),
+        };
+
+        let subscriber = Arc::new(Box::new(NamingChangeSubscriber) as Box<dyn Subscriber>);
+
+        super::register(subscriber.clone());
+
+        super::post(Box::new(event));
+
+        let one_millis = time::Duration::from_secs(1);
+        thread::sleep(one_millis);
+
+        super::unregister(subscriber);
+
+        let event = NamingChangeEvent {
+            message: "unregister".to_owned(),
+        };
+        super::post(Box::new(event));
+
+        let one_millis = time::Duration::from_secs(1);
+        thread::sleep(one_millis);
+    }
 }

src/naming/grpc/grpc_service.rs

@@ -399,6 +399,7 @@ mod tests {
     use super::*;
 
     #[test]
+    #[ignore]
     pub fn test_check_server() {
         let collector = tracing_subscriber::fmt()
             .with_thread_names(true)
@@ -416,6 +417,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore]
     pub fn test_grpc_server_builder() {
         let collector = tracing_subscriber::fmt()
             .with_thread_names(true)