refactor: address feedback for multiple listener versions

Implement comprehensive improvements:

- Replace HashMap<String, Vec<ListenerInfo>> with MultiMap for cleaner API
- Move default config assignment inside ListenerManager::new() constructor
- Change listener exit logging from warn to info level
- Merge redundant version logging into single informative message
- Fix unnecessary double cloning of listener_name
- Use Vec::drain for efficient batch removal of old versions
- Remove unimplemented TimeBasedOnly cleanup policy to avoid incorrect behavior
- Remove redundant Hybrid cleanup policy, keeping only CountBasedOnly

Signed-off-by: Eeshu-Yadav <eeshuyadav123@gmail.com>

Author: Eeshu-Yadav

orion-lib/src/lib.rs

@@ -140,11 +140,7 @@ pub async fn start_listener_manager(
         configuration_receivers;
 
     tracing::debug!("listeners manager starting");
-    let mgr = ListenersManager::new(
-        listener_configuration_receiver,
-        route_configuration_receiver,
-        listeners::listeners_manager::ListenerManagerConfig::default(),
-    );
+    let mgr = ListenersManager::new(listener_configuration_receiver, route_configuration_receiver);
     mgr.start(ct).await.map_err(|err| {
         tracing::warn!(error = %err, "listeners manager exited with error");
         err

orion-lib/src/listeners/listeners_manager.rs

@@ -15,7 +15,7 @@
 //
 //
 
-use std::collections::HashMap;
+use multimap::MultiMap;
 use std::time::Duration;
 
 use tokio::sync::{broadcast, mpsc};
@@ -55,8 +55,6 @@ pub struct ListenerManagerConfig {
 #[derive(Debug, Clone)]
 pub enum CleanupPolicy {
     CountBasedOnly(usize),
-    TimeBasedOnly(Duration),
-    Hybrid { timeout: Duration, max_count: usize },
 }
 
 impl Default for ListenerManagerConfig {
@@ -92,6 +90,13 @@ impl ListenersManager {
     pub fn new(
         listener_configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
         route_configuration_channel: mpsc::Receiver<RouteConfigurationChange>,
+    ) -> Self {
+        Self::with_config(listener_configuration_channel, route_configuration_channel, ListenerManagerConfig::default())
+    }
+
+    pub fn with_config(
+        listener_configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
+        route_configuration_channel: mpsc::Receiver<RouteConfigurationChange>,
         config: ListenerManagerConfig,
     ) -> Self {
         ListenersManager {
@@ -170,7 +175,6 @@ impl ListenersManager {
 
         let listener_info = ListenerInfo::new(join_handle, listener_conf, version);
         self.listener_handles.insert(listener_name.clone(), listener_info);
-
         let version_count = self.listener_handles.get_vec(&listener_name).map(|v| v.len()).unwrap_or(0);
         info!("Started version {} of listener {} ({} total active version(s))", version, listener_name, version_count);
 
@@ -194,48 +198,33 @@ impl ListenersManager {
     }
 
     fn cleanup_old_versions(&mut self, listener_name: &str) {
-        if let Some(versions) = self.listener_handles.get_mut(listener_name) {
+        if let Some(mut versions) = self.listener_handles.remove(listener_name) {
             let original_count = versions.len();
 
             match &self.config.cleanup_policy {
                 CleanupPolicy::CountBasedOnly(max_count) => {
                     if versions.len() > *max_count {
                         let to_remove = versions.len() - max_count;
-                        for _ in 0..to_remove {
-                            let old = versions.remove(0);
+                        let removed = versions.drain(0..to_remove).collect::<Vec<_>>();
+                        for old in removed {
                             info!("Cleaning up old listener {} version {} (count limit)", listener_name, old.version);
                         }
                     }
                 },
-                CleanupPolicy::TimeBasedOnly(_timeout) => {
-                    // TODO: Implement time-based cleanup when we have connection tracking
-                    // For now, behave like count-based with default limit
-                    if versions.len() > self.config.max_versions_per_listener {
-                        let to_remove = versions.len() - self.config.max_versions_per_listener;
-                        for _ in 0..to_remove {
-                            let old = versions.remove(0);
-                            info!("Cleaning up old listener {} version {} (time limit)", listener_name, old.version);
-                        }
-                    }
-                },
-                CleanupPolicy::Hybrid { max_count, .. } => {
-                    if versions.len() > *max_count {
-                        let to_remove = versions.len() - max_count;
-                        for _ in 0..to_remove {
-                            let old = versions.remove(0);
-                            info!("Cleaning up old listener {} version {} (hybrid limit)", listener_name, old.version);
-                        }
-                    }
-                },
             }
 
-            let cleaned_count = original_count - versions.len();
-            if cleaned_count > 0 {
+            // Re-insert the remaining versions
+            for version in versions {
+                self.listener_handles.insert(listener_name.to_string(), version);
+            }
+
+            let remaining_count = self.listener_handles.get_vec(listener_name).map(|v| v.len()).unwrap_or(0);
+            if original_count != remaining_count {
                 info!(
-                    "Cleaned up {} old versions of listener {}, {} versions remaining",
-                    cleaned_count,
+                    "Cleaned up {} old version(s) of listener {}, {} remaining",
+                    original_count - remaining_count,
                     listener_name,
-                    versions.len()
+                    remaining_count
                 );
             }
         }
@@ -274,8 +263,7 @@ mod tests {
 
         let (_conf_tx, conf_rx) = mpsc::channel(chan);
         let (_route_tx, route_rx) = mpsc::channel(chan);
-        let config = ListenerManagerConfig::default();
-        let mut man = ListenersManager::new(conf_rx, route_rx, config);
+        let mut man = ListenersManager::new(conf_rx, route_rx);
 
         let (routeb_tx1, routeb_rx) = broadcast::channel(chan);
         let (_secb_tx1, secb_rx) = broadcast::channel(chan);
@@ -318,8 +306,7 @@ mod tests {
 
         let (_conf_tx, conf_rx) = mpsc::channel(chan);
         let (_route_tx, route_rx) = mpsc::channel(chan);
-        let config = ListenerManagerConfig::default();
-        let mut man = ListenersManager::new(conf_rx, route_rx, config);
+        let mut man = ListenersManager::new(conf_rx, route_rx);
 
         let (routeb_tx1, routeb_rx) = broadcast::channel(chan);
         let (secb_tx1, secb_rx) = broadcast::channel(chan);
@@ -366,7 +353,7 @@ mod tests {
             cleanup_policy: CleanupPolicy::CountBasedOnly(2),
             cleanup_interval: Duration::from_secs(60),
         };
-        let mut man = ListenersManager::new(conf_rx, route_rx, config);
+        let mut man = ListenersManager::with_config(conf_rx, route_rx, config);
 
         let (routeb_tx1, routeb_rx) = broadcast::channel(chan);
         let (_secb_tx1, secb_rx) = broadcast::channel(chan);
@@ -399,7 +386,7 @@ mod tests {
         assert!(routeb_tx3.send(RouteConfigurationChange::Removed("n/a".into())).is_ok());
 
         // Should only have 2 versions due to cleanup policy (max_count: 2)
-        assert_eq!(man.listener_handles.get(name).unwrap().len(), 2);
+        assert_eq!(man.listener_handles.get_vec(name).unwrap().len(), 2);
 
         man.stop_listener(name).unwrap();
 
@@ -421,7 +408,7 @@ mod tests {
             cleanup_policy: CleanupPolicy::CountBasedOnly(3),
             cleanup_interval: Duration::from_secs(60),
         };
-        let mut man = ListenersManager::new(conf_rx, route_rx, config);
+        let mut man = ListenersManager::with_config(conf_rx, route_rx, config);
 
         // Add 5 listeners, should only keep 3 due to cleanup policy
         for i in 1..=5 {
@@ -434,10 +421,10 @@ mod tests {
         }
 
         // Should only have 3 versions due to cleanup policy
-        assert_eq!(man.listener_handles.get(name).unwrap().len(), 3);
+        assert_eq!(man.listener_handles.get_vec(name).unwrap().len(), 3);
 
         man.stop_listener(name).unwrap();
-        assert!(man.listener_handles.get(name).is_none());
+        assert!(man.listener_handles.get_vec(name).is_none());
 
         tokio::task::yield_now().await;
     }