Address Copilot feedback: remove dead code, make buffers configurable, add TODOs

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

Author: Eeshu-Yadav

orion-internal/src/connection.rs

@@ -42,15 +42,17 @@ pub struct InternalListenerHandle {
     pub name: String,
     pub connection_sender: mpsc::UnboundedSender<InternalConnectionPair>,
     listener_ref: Weak<()>,
+    buffer_size_kb: Option<u32>,
 }
 
 impl InternalListenerHandle {
     pub fn new(
         name: String,
         connection_sender: mpsc::UnboundedSender<InternalConnectionPair>,
         listener_ref: Weak<()>,
+        buffer_size_kb: Option<u32>,
     ) -> Self {
-        Self { name, connection_sender, listener_ref }
+        Self { name, connection_sender, listener_ref, buffer_size_kb }
     }
 
     pub fn is_alive(&self) -> bool {
@@ -64,7 +66,7 @@ impl InternalListenerHandle {
 
         let metadata = InternalConnectionMetadata {
             listener_name: self.name.clone(),
-            buffer_size_kb: None,
+            buffer_size_kb: self.buffer_size_kb,
             created_at: Instant::now(),
             endpoint_id,
         };
@@ -185,12 +187,13 @@ impl InternalConnectionFactory {
     pub async fn register_listener(
         &self,
         name: String,
+        buffer_size_kb: Option<u32>,
     ) -> Result<(InternalListenerHandle, mpsc::UnboundedReceiver<InternalConnectionPair>, Arc<()>)> {
         let (connection_tx, connection_rx) = mpsc::unbounded_channel();
         let listener_ref = Arc::new(());
         let weak_ref = Arc::downgrade(&listener_ref);
 
-        let handle = InternalListenerHandle::new(name.clone(), connection_tx, weak_ref);
+        let handle = InternalListenerHandle::new(name.clone(), connection_tx, weak_ref, buffer_size_kb);
 
         let mut listeners = self.listeners.write().await;
 
@@ -337,8 +340,12 @@ impl AsyncWrite for InternalStreamWrapper {
     }
 }
 
+const DEFAULT_BUFFER_SIZE: usize = 8192;
+
 fn create_internal_connection_pair(metadata: InternalConnectionMetadata) -> (Arc<InternalStream>, Arc<InternalStream>) {
-    let (upstream_io, downstream_io) = tokio::io::duplex(1024);
+    let buffer_size = metadata.buffer_size_kb.map(|kb| (kb as usize) * 1024).unwrap_or(DEFAULT_BUFFER_SIZE);
+
+    let (upstream_io, downstream_io) = tokio::io::duplex(buffer_size);
 
     let upstream = Arc::new(InternalStream::new(metadata.clone(), upstream_io));
     let downstream = Arc::new(InternalStream::new(metadata, downstream_io));
@@ -368,22 +375,22 @@ mod tests {
     async fn test_listener_registration() {
         let factory = InternalConnectionFactory::new();
 
-        let result = factory.register_listener("test_listener".to_string()).await;
+        let result = factory.register_listener("test_listener".to_string(), None).await;
         assert!(result.is_ok());
         let (_handle, _rx, _listener_ref) = result.unwrap();
 
         let stats = factory.get_stats().await;
         assert_eq!(stats.active_listeners, 1);
 
-        let result = factory.register_listener("test_listener".to_string()).await;
+        let result = factory.register_listener("test_listener".to_string(), None).await;
         assert!(result.is_err());
     }
 
     #[tokio::test]
     async fn test_listener_unregistration() {
         let factory = InternalConnectionFactory::new();
 
-        let (_handle, _rx, _listener_ref) = factory.register_listener("test_listener".to_string()).await.unwrap();
+        let (_handle, _rx, _listener_ref) = factory.register_listener("test_listener".to_string(), None).await.unwrap();
         let result = factory.unregister_listener("test_listener").await;
         assert!(result.is_ok());
 
@@ -406,7 +413,7 @@ mod tests {
     async fn test_listener_lifecycle() {
         let factory = InternalConnectionFactory::new();
 
-        let (handle, _rx, _listener_ref) = factory.register_listener("test_listener".to_string()).await.unwrap();
+        let (handle, _rx, _listener_ref) = factory.register_listener("test_listener".to_string(), None).await.unwrap();
 
         assert!(factory.is_listener_active("test_listener").await);
         assert!(handle.is_alive());
@@ -423,8 +430,8 @@ mod tests {
         let listeners = factory.list_listeners().await;
         assert!(listeners.is_empty());
 
-        let (_handle1, _rx1, _listener_ref1) = factory.register_listener("listener1".to_string()).await.unwrap();
-        let (_handle2, _rx2, _listener_ref2) = factory.register_listener("listener2".to_string()).await.unwrap();
+        let (_handle1, _rx1, _listener_ref1) = factory.register_listener("listener1".to_string(), None).await.unwrap();
+        let (_handle2, _rx2, _listener_ref2) = factory.register_listener("listener2".to_string(), None).await.unwrap();
 
         let listeners = factory.list_listeners().await;
         assert_eq!(listeners.len(), 2);
@@ -438,4 +445,27 @@ mod tests {
         let stats = factory.get_stats().await;
         assert_eq!(stats.max_pooled_connections, 0);
     }
+
+    #[tokio::test]
+    async fn test_buffer_size_configuration() {
+        let metadata_default = InternalConnectionMetadata {
+            listener_name: "test".to_string(),
+            buffer_size_kb: None,
+            created_at: Instant::now(),
+            endpoint_id: None,
+        };
+        let (upstream, downstream) = create_internal_connection_pair(metadata_default);
+        assert!(upstream.is_active());
+        assert!(downstream.is_active());
+
+        let metadata_custom = InternalConnectionMetadata {
+            listener_name: "test".to_string(),
+            buffer_size_kb: Some(4),
+            created_at: Instant::now(),
+            endpoint_id: None,
+        };
+        let (upstream_custom, downstream_custom) = create_internal_connection_pair(metadata_custom);
+        assert!(upstream_custom.is_active());
+        assert!(downstream_custom.is_active());
+    }
 }

orion-internal/tests/integration_tests.rs

@@ -23,7 +23,7 @@ async fn test_complete_internal_connection_flow() {
 
     let listener_name = "test_integration_listener_global";
     let (_handle, mut connection_receiver, _listener_ref) =
-        factory.register_listener(listener_name.to_string()).await.expect("Failed to register listener");
+        factory.register_listener(listener_name.to_string(), None).await.expect("Failed to register listener");
 
     assert!(factory.is_listener_active(listener_name).await);
     let listeners = factory.list_listeners().await;
@@ -67,7 +67,7 @@ async fn test_connection_pooling() {
 
     let listener_name = "test_pooling_listener_global";
     let (_handle, mut connection_receiver, _listener_ref) =
-        factory.register_listener(listener_name.to_string()).await.expect("Failed to register listener");
+        factory.register_listener(listener_name.to_string(), None).await.expect("Failed to register listener");
 
     let connector = InternalChannelConnector::new(listener_name.to_string(), "test_cluster", None);
 
@@ -94,11 +94,11 @@ async fn test_error_scenarios() {
     assert!(result.is_err());
 
     let listener_name = "test_error_listener_global";
-    let result1 = factory.register_listener(listener_name.to_string()).await;
+    let result1 = factory.register_listener(listener_name.to_string(), None).await;
     assert!(result1.is_ok());
     let (_handle1, _rx1, _listener_ref1) = result1.unwrap();
 
-    let result2 = factory.register_listener(listener_name.to_string()).await;
+    let result2 = factory.register_listener(listener_name.to_string(), None).await;
     assert!(result2.is_err());
 
     let result = factory.unregister_listener("non_existent").await;
@@ -124,7 +124,7 @@ async fn test_global_factory() {
     let factory = global_internal_connection_factory();
 
     let listener_name = "test_global_listener";
-    let (_handle, _rx, _listener_ref) = factory.register_listener(listener_name.to_string()).await.unwrap();
+    let (_handle, _rx, _listener_ref) = factory.register_listener(listener_name.to_string(), None).await.unwrap();
 
     assert!(factory.is_listener_active(listener_name).await);
 
@@ -141,7 +141,7 @@ async fn test_statistics_and_monitoring() {
     let listener1 = "test_stats_listener1_global";
 
     let (_handle1, _rx1, _listener_ref1) =
-        factory.register_listener(listener1.to_string()).await.expect("Failed to register listener1");
+        factory.register_listener(listener1.to_string(), None).await.expect("Failed to register listener1");
 
     let stats = factory.get_stats().await;
     assert!(stats.active_listeners >= 1);
@@ -172,7 +172,7 @@ async fn test_cluster_helpers() {
 
     let factory = global_internal_connection_factory();
     let (_handle, _rx, _listener_ref) =
-        factory.register_listener("test_cluster_helpers_listener".to_string()).await.unwrap();
+        factory.register_listener("test_cluster_helpers_listener".to_string(), None).await.unwrap();
 
     assert!(is_internal_listener_available("test_cluster_helpers_listener").await);
 
@@ -181,3 +181,29 @@ async fn test_cluster_helpers() {
 
     factory.unregister_listener("test_cluster_helpers_listener").await.unwrap();
 }
+
+#[tokio::test]
+async fn test_buffer_size_configuration() {
+    let factory = global_internal_connection_factory();
+
+    let listener_name = "test_buffer_size_listener";
+    let buffer_size_kb = Some(8);
+
+    let (_handle, mut connection_receiver, _listener_ref) =
+        factory.register_listener(listener_name.to_string(), buffer_size_kb).await.unwrap();
+
+    let connector = InternalChannelConnector::new(listener_name.to_string(), "test_cluster", None);
+
+    let connection_future = connector.connect();
+    let listener_future = connection_receiver.recv();
+    let (_cluster_conn, listener_pair) = tokio::join!(connection_future, listener_future);
+
+    assert!(listener_pair.is_some());
+    let connection_pair = listener_pair.unwrap();
+
+    let metadata = connection_pair.downstream.metadata();
+    assert_eq!(metadata.buffer_size_kb, Some(8));
+    assert_eq!(metadata.listener_name, listener_name);
+
+    factory.unregister_listener(listener_name).await.unwrap();
+}

orion-lib/src/lib.rs

@@ -44,7 +44,7 @@ pub use clusters::{
     load_assignment::PartialClusterLoadAssignment,
     ClusterLoadAssignmentBuilder,
 };
-pub use listeners::listener::ListenerFactory;
+pub use listeners::listener::{init_internal_worker_pool, ListenerFactory};
 pub use listeners_manager::{ListenerConfigurationChange, ListenersManager, RouteConfigurationChange};
 pub use orion_configuration::config::network_filters::http_connection_manager::RouteConfiguration;
 use orion_configuration::config::{

orion-lib/src/listeners/listener.rs

@@ -84,7 +84,6 @@ impl InternalConnectionWorkerPool {
                 while let Some(task) = receiver.recv().await {
                     if let Err(e) =
                         handle_internal_connection_static(task.listener_name, task.connection_pair, task.filter_chains)
-                            .await
                     {
                         warn!("Error handling internal connection task: {}", e);
                     }
@@ -112,8 +111,19 @@ impl InternalConnectionWorkerPool {
 
 static INTERNAL_WORKER_POOL: std::sync::OnceLock<InternalConnectionWorkerPool> = std::sync::OnceLock::new();
 
+pub fn init_internal_worker_pool(num_workers: Option<usize>) {
+    let workers = num_workers
+        .or_else(|| std::env::var("ORION_INTERNAL_WORKER_POOL_SIZE").ok().and_then(|s| s.parse::<usize>().ok()))
+        .unwrap_or(4);
+    let _ = INTERNAL_WORKER_POOL.set(InternalConnectionWorkerPool::new(workers));
+}
+
 fn get_internal_worker_pool() -> &'static InternalConnectionWorkerPool {
-    INTERNAL_WORKER_POOL.get_or_init(|| InternalConnectionWorkerPool::new(4)) // 4 workers by default
+    INTERNAL_WORKER_POOL.get_or_init(|| {
+        let workers =
+            std::env::var("ORION_INTERNAL_WORKER_POOL_SIZE").ok().and_then(|s| s.parse::<usize>().ok()).unwrap_or(4);
+        InternalConnectionWorkerPool::new(workers)
+    })
 }
 
 #[derive(Debug, Clone)]
@@ -135,7 +145,6 @@ enum ListenerAddress {
 
 #[derive(Debug, Clone)]
 struct InternalListenerConfig {
-    #[allow(dead_code)]
     buffer_size_kb: Option<u32>,
 }
 #[derive(Debug, Clone)]
@@ -395,7 +404,7 @@ impl Listener {
 
     async fn run_internal_listener(
         name: &'static str,
-        _internal_config: InternalListenerConfig,
+        internal_config: InternalListenerConfig,
         filter_chains: HashMap<FilterChainMatch, FilterchainType>,
         _with_tls_inspector: bool,
         _with_tlv_listener_filter: bool,
@@ -408,13 +417,14 @@ impl Listener {
         let filter_chains = Arc::new(filter_chains);
         let factory = global_internal_connection_factory();
 
-        let (_handle, mut connection_receiver, _listener_ref) = match factory.register_listener(name.to_owned()).await {
-            Ok(result) => result,
-            Err(e) => {
-                error!("Failed to register internal listener '{}': {}", name, e);
-                return e;
-            },
-        };
+        let (_handle, mut connection_receiver, _listener_ref) =
+            match factory.register_listener(name.to_owned(), internal_config.buffer_size_kb).await {
+                Ok(result) => result,
+                Err(e) => {
+                    error!("Failed to register internal listener '{}': {}", name, e);
+                    return e;
+                },
+            };
 
         info!("Internal listener '{}' registered to connection factory", name);
 
@@ -983,7 +993,7 @@ filter_chains:
     }
 }
 
-async fn handle_internal_connection_static(
+fn handle_internal_connection_static(
     listener_name: String,
     connection_pair: orion_internal::InternalConnectionPair,
     filter_chains: Arc<HashMap<FilterChainMatch, FilterchainType>>,
@@ -1004,17 +1014,19 @@ async fn handle_internal_connection_static(
         return Err(crate::Error::new("No matching filter chain"));
     };
 
-    let _downstream_stream = connection_pair.downstream;
+    let downstream_stream = connection_pair.downstream;
 
     match &filter_chain.handler {
         crate::listeners::filterchain::ConnectionHandler::Http(_http_manager) => {
             info!("Processing internal connection through HTTP filter chain");
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            // TODO: Implement HTTP connection processing
+            let _ = downstream_stream;
             Ok(())
         },
         crate::listeners::filterchain::ConnectionHandler::Tcp(_tcp_proxy) => {
             info!("Processing internal connection through TCP filter chain");
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            // TODO: Implement TCP connection processing
+            let _ = downstream_stream;
             Ok(())
         },
     }