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,14 @@ impl AsyncWrite for InternalStreamWrapper {
     }
 }
 
+const DEFAULT_BUFFER_SIZE: usize = 1024;
+
 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 +377,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 +415,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 +432,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 +447,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/listeners/listener.rs

@@ -135,7 +135,6 @@ enum ListenerAddress {
 
 #[derive(Debug, Clone)]
 struct InternalListenerConfig {
-    #[allow(dead_code)]
     buffer_size_kb: Option<u32>,
 }
 #[derive(Debug, Clone)]
@@ -395,7 +394,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,7 +407,10 @@ 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 {
+        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);
@@ -1004,17 +1006,23 @@ async fn handle_internal_connection_static(
         return Err(crate::Error::new("No matching filter chain"));
     };
 
+    // TODO: Implement actual processing of the downstream stream for internal connections.
+    // The downstream stream should be passed to the filter chain handler for proper processing.
     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 actual HTTP connection processing through the filter chain
+            // This should involve setting up HTTP/1.1 or HTTP/2 protocol handling,
+            // routing requests through the configured route table, and applying filters.
             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 actual TCP connection processing through the filter chain
+            // This should involve proxying the TCP connection to the configured upstream
+            // cluster and handling bidirectional data transfer.
             Ok(())
         },
     }