feat: Extract internal connection infrastructure to orion-internal crate

Moved all internal connection related code (factories, connectors, streams, tests)
from orion-lib to dedicated orion-internal crate.

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

Author: Eeshu-Yadav

Cargo.lock

@@ -6,6 +6,7 @@ members = [
   "orion-error",
   "orion-http-header",
   "orion-interner",
+  "orion-internal",
   "orion-format",
   "orion-lib",
   "orion-metrics",
@@ -35,6 +36,7 @@ orion-error          = { path = "orion-error" }
 orion-format         = { path = "orion-format" }
 orion-http-header    = { path = "orion-http-header" }
 orion-interner       = { path = "orion-interner" }
+orion-internal       = { path = "orion-internal" }
 orion-lib            = { path = "orion-lib" }
 orion-metrics        = { path = "orion-metrics" }
 orion-tracing        = { path = "orion-tracing" }

Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "orion-internal"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+description = "Internal connection and internal listener infrastructure for Orion proxy"
+
+[dependencies]
+tokio = { version = "1.42.0", features = ["full"] }
+tracing = "0.1.41"
+orion-error = { path = "../orion-error" }
+orion-configuration = { path = "../orion-configuration" }
+compact_str = "0.8.0"
+
+[dev-dependencies]
+tokio-test = "0.4"

orion-internal/Cargo.toml

@@ -27,8 +27,7 @@ use tokio::{
     time::Instant,
 };
 
-use super::AsyncStream;
-use crate::{Error, Result};
+pub use crate::{Error, Result};
 
 #[derive(Debug, Clone)]
 pub struct InternalConnectionMetadata {
@@ -213,7 +212,11 @@ impl InternalConnectionFactory {
         Ok(())
     }
 
-    pub async fn connect_to_listener(&self, name: &str, endpoint_id: Option<String>) -> Result<AsyncStream> {
+    pub async fn connect_to_listener(
+        &self,
+        name: &str,
+        endpoint_id: Option<String>,
+    ) -> Result<Box<InternalStreamWrapper>> {
         let listeners = self.listeners.read().await;
         let handle = listeners.get(name).ok_or_else(|| Error::new(format!("Internal listener '{name}' not found")))?;
 

orion-lib/src/transport/internal_connection.rs renamed to orion-internal/src/connection.rs

@@ -15,8 +15,7 @@
 //
 //
 
-use super::{global_internal_connection_factory, AsyncStream};
-use crate::{Error, Result};
+use crate::{connection::global_internal_connection_factory, Error, InternalStreamWrapper, Result};
 
 #[derive(Debug, Clone)]
 pub struct InternalClusterConnector {
@@ -37,7 +36,7 @@ impl InternalClusterConnector {
         self.endpoint_id.as_deref()
     }
 
-    pub async fn connect(&self) -> Result<AsyncStream> {
+    pub async fn connect(&self) -> Result<Box<InternalStreamWrapper>> {
         let factory = global_internal_connection_factory();
 
         if !factory.is_listener_active(&self.listener_name).await {
@@ -94,7 +93,7 @@ impl InternalChannelConnector {
 }
 
 pub struct InternalChannel {
-    pub stream: AsyncStream,
+    pub stream: Box<InternalStreamWrapper>,
     pub cluster_name: &'static str,
     pub listener_name: String,
     pub endpoint_id: Option<String>,
@@ -116,6 +115,7 @@ impl InternalChannel {
 
 pub mod cluster_helpers {
     use super::{global_internal_connection_factory, InternalChannelConnector};
+    use crate::InternalConnectionStats;
     use orion_configuration::config::cluster::InternalEndpointAddress;
 
     pub fn create_internal_connector(
@@ -134,7 +134,7 @@ pub mod cluster_helpers {
         factory.is_listener_active(listener_name).await
     }
 
-    pub async fn get_internal_connection_stats() -> crate::transport::InternalConnectionStats {
+    pub async fn get_internal_connection_stats() -> InternalConnectionStats {
         let factory = global_internal_connection_factory();
         factory.get_stats().await
     }

orion-lib/src/transport/internal_cluster_connector.rs renamed to orion-internal/src/connector.rs

@@ -0,0 +1,49 @@
+// Copyright 2025 The kmesh Authors
+//
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//
+
+use std::net::SocketAddr;
+
+/// Reserved internal address used as the peer address for in-process (internal) connections.
+/// Chosen from the loopback range (127.255.255.254:65534) to avoid conflicts with real network traffic.
+/// This address clearly identifies internal connections in logs and debugging.
+pub const INTERNAL_PEER_ADDR: SocketAddr =
+    SocketAddr::V4(std::net::SocketAddrV4::new(std::net::Ipv4Addr::new(127, 255, 255, 254), 65534));
+
+/// Reserved internal address used as the local address for in-process (internal) connections.
+/// Chosen from the loopback range (127.255.255.255:65535) to avoid conflicts with real network traffic.
+/// This address clearly identifies internal connections in logs and debugging.
+pub const INTERNAL_LOCAL_ADDR: SocketAddr =
+    SocketAddr::V4(std::net::SocketAddrV4::new(std::net::Ipv4Addr::new(127, 255, 255, 255), 65535));
+
+/// Check if a socket address is an internal connection address
+pub fn is_internal_address(addr: &SocketAddr) -> bool {
+    *addr == INTERNAL_PEER_ADDR || *addr == INTERNAL_LOCAL_ADDR
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_internal_addresses() {
+        assert!(is_internal_address(&INTERNAL_PEER_ADDR));
+        assert!(is_internal_address(&INTERNAL_LOCAL_ADDR));
+
+        let normal_addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
+        assert!(!is_internal_address(&normal_addr));
+    }
+}

orion-internal/src/filter_state.rs

@@ -0,0 +1,39 @@
+// Copyright 2025 The kmesh Authors
+//
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//
+
+//! # orion-internal
+//!
+//! Internal connection and internal listener infrastructure for Orion proxy.
+//! This crate provides the core functionality for in-process communication between
+//! different components of the proxy, particularly for waypoint proxy capabilities
+//! in ambient service mesh scenarios.
+
+pub mod connection;
+pub mod connector;
+pub mod filter_state;
+
+// Re-export commonly used types
+pub use connection::{
+    global_internal_connection_factory, InternalConnectionFactory, InternalConnectionMetadata, InternalConnectionPair,
+    InternalConnectionStats, InternalListenerHandle, InternalStream, InternalStreamWrapper,
+};
+pub use connector::{cluster_helpers, InternalChannel, InternalChannelConnector, InternalClusterConnector};
+pub use filter_state::{is_internal_address, INTERNAL_LOCAL_ADDR, INTERNAL_PEER_ADDR};
+
+// Re-export error types
+pub type Error = orion_error::Error;
+pub type Result<T> = ::core::result::Result<T, Error>;

orion-internal/src/lib.rs

@@ -15,7 +15,7 @@
 //
 //
 
-use orion_lib::{global_internal_connection_factory, InternalChannelConnector};
+use orion_internal::{cluster_helpers::*, global_internal_connection_factory, InternalChannelConnector};
 
 #[tokio::test]
 async fn test_complete_internal_connection_flow() {
@@ -155,22 +155,29 @@ async fn test_statistics_and_monitoring() {
 #[tokio::test]
 async fn test_cluster_helpers() {
     use orion_configuration::config::cluster::InternalEndpointAddress;
-    use orion_lib::cluster_helpers::*;
 
     let internal_addr = InternalEndpointAddress {
-        server_listener_name: "test_listener".to_string().into(),
+        server_listener_name: "test_cluster_helpers_listener".to_string().into(),
         endpoint_id: Some("endpoint1".to_string().into()),
     };
 
     let connector = create_internal_connector(&internal_addr, "test_cluster");
     assert_eq!(connector.cluster_name(), "test_cluster");
-    assert_eq!(connector.listener_name(), "test_listener");
+    assert_eq!(connector.listener_name(), "test_cluster_helpers_listener");
 
-    assert!(!is_internal_listener_available("non_existent").await);
+    assert!(!is_internal_listener_available("non_existent_listener_xyz").await);
 
     let stats = get_internal_connection_stats().await;
     assert_eq!(stats.max_pooled_connections, 0);
 
+    let factory = global_internal_connection_factory();
+    let (_handle, _rx, _listener_ref) =
+        factory.register_listener("test_cluster_helpers_listener".to_string()).await.unwrap();
+
+    assert!(is_internal_listener_available("test_cluster_helpers_listener").await);
+
     let listeners = list_internal_listeners().await;
-    assert!(listeners.is_empty());
+    assert!(listeners.contains(&"test_cluster_helpers_listener".to_string()));
+
+    factory.unregister_listener("test_cluster_helpers_listener").await.unwrap();
 }

orion-lib/tests/internal_connection_integration.rs renamed to orion-internal/tests/integration_tests.rs

@@ -37,6 +37,7 @@ orion-error.workspace = true
 orion-format.workspace = true
 orion-http-header.workspace = true
 orion-interner.workspace = true
+orion-internal.workspace = true
 orion-metrics.workspace = true
 orion-tracing.workspace = true
 orion-xds.workspace = true

orion-lib/Cargo.toml

@@ -54,11 +54,11 @@ use orion_configuration::config::{
     Bootstrap, Cluster, Listener as ListenerConfig,
 };
 pub use secrets::SecretManager;
-pub use transport::internal_cluster_connector::cluster_helpers;
 pub(crate) use transport::AsyncStream;
-pub use transport::{
-    global_internal_connection_factory, InternalChannelConnector, InternalConnectionFactory, InternalConnectionPair,
-    InternalConnectionStats, InternalListenerHandle,
+
+pub use orion_internal::{
+    cluster_helpers, global_internal_connection_factory, InternalChannelConnector, InternalConnectionFactory,
+    InternalConnectionPair, InternalConnectionStats, InternalListenerHandle,
 };
 
 pub type Error = orion_error::Error;