Repository: krojew/cdrs-tokio (language: rust)

Issue:
Add support for HTTP proxy



Discussion / hints:
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.


Fix patch:
diff --git a/cdrs-tokio/Cargo.toml b/cdrs-tokio/Cargo.toml
index a61f7c3..c0b9f67 100644
--- a/cdrs-tokio/Cargo.toml
+++ b/cdrs-tokio/Cargo.toml
@@ -15,6 +15,7 @@ license = "MIT/Apache-2.0"
 rust-tls = ["rustls", "tokio-rustls", "webpki"]
 e2e-tests = []
 derive = ["cdrs-tokio-helpers-derive"]
+http-proxy = ["async-http-proxy"]
 
 [dependencies]
 arc-swap ="1.4.0"
@@ -37,6 +38,11 @@ tracing = "0.1.35"
 uuid = "1.0.0"
 webpki = { version = "0.22.0", optional = true }
 
+[dependencies.async-http-proxy]
+version = "1.2.5"
+optional = true
+features = ["runtime-tokio", "basic-auth"]
+
 [dependencies.rustls]
 version = "0.20.6"
 optional = true
diff --git a/cdrs-tokio/examples/generic_connection.rs b/cdrs-tokio/examples/generic_connection.rs
index 4c797a2..504eadc 100644
--- a/cdrs-tokio/examples/generic_connection.rs
+++ b/cdrs-tokio/examples/generic_connection.rs
@@ -117,6 +117,8 @@ impl VirtualConnectionManager {
                 DEFAULT_TRANSPORT_BUFFER_SIZE,
                 true,
                 config.version,
+                #[cfg(feature = "http-proxy")]
+                None,
             ),
             mask: config.mask,
             actual: config.actual,
diff --git a/cdrs-tokio/src/cluster.rs b/cdrs-tokio/src/cluster.rs
index 49beee9..116623a 100644
--- a/cdrs-tokio/src/cluster.rs
+++ b/cdrs-tokio/src/cluster.rs
@@ -1,4 +1,6 @@
 pub(crate) use self::cluster_metadata_manager::ClusterMetadataManager;
+#[cfg(feature = "http-proxy")]
+pub use self::config_proxy::{HttpProxyConfig, HttpProxyConfigBuilder};
 #[cfg(feature = "rust-tls")]
 pub use self::config_rustls::{NodeRustlsConfig, NodeRustlsConfigBuilder};
 pub use self::config_tcp::{NodeTcpConfig, NodeTcpConfigBuilder};
@@ -22,6 +24,8 @@ use cassandra_protocol::frame::Version;
 use std::sync::Arc;
 
 mod cluster_metadata_manager;
+#[cfg(feature = "http-proxy")]
+mod config_proxy;
 #[cfg(feature = "rust-tls")]
 mod config_rustls;
 mod config_tcp;
diff --git a/cdrs-tokio/src/cluster/config_proxy.rs b/cdrs-tokio/src/cluster/config_proxy.rs
new file mode 100644
index 0000000..262e916
--- /dev/null
+++ b/cdrs-tokio/src/cluster/config_proxy.rs
@@ -0,0 +1,41 @@
+#[derive(Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]
+pub(crate) struct HttpBasicAuth {
+    pub(crate) username: String,
+    pub(crate) password: String,
+}
+
+#[derive(Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]
+pub struct HttpProxyConfig {
+    pub(crate) address: String,
+    pub(crate) basic_auth: Option<HttpBasicAuth>,
+}
+
+#[derive(Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]
+pub struct HttpProxyConfigBuilder {
+    address: String,
+    basic_auth: Option<HttpBasicAuth>,
+}
+
+impl HttpProxyConfigBuilder {
+    /// Creates a new proxy configuration builder with given proxy address.
+    pub fn new(address: String) -> Self {
+        Self {
+            address,
+            basic_auth: None,
+        }
+    }
+
+    /// Adds HTTP basic Auth.
+    pub fn with_basic_auth(mut self, username: String, password: String) -> Self {
+        self.basic_auth = Some(HttpBasicAuth { password, username });
+        self
+    }
+
+    /// Build the resulting configuration.
+    pub fn build(self) -> HttpProxyConfig {
+        HttpProxyConfig {
+            basic_auth: self.basic_auth,
+            address: self.address,
+        }
+    }
+}
diff --git a/cdrs-tokio/src/cluster/config_rustls.rs b/cdrs-tokio/src/cluster/config_rustls.rs
index 249c872..3ce29e2 100644
--- a/cdrs-tokio/src/cluster/config_rustls.rs
+++ b/cdrs-tokio/src/cluster/config_rustls.rs
@@ -1,24 +1,28 @@
+use cassandra_protocol::authenticators::{NoneAuthenticatorProvider, SaslAuthenticatorProvider};
+use cassandra_protocol::error::Result;
+use cassandra_protocol::frame::Version;
 use derivative::Derivative;
+use rustls::ServerName;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+#[cfg(feature = "http-proxy")]
+use crate::cluster::HttpProxyConfig;
 use crate::cluster::NodeAddress;
-use cassandra_protocol::authenticators::{NoneAuthenticatorProvider, SaslAuthenticatorProvider};
-use cassandra_protocol::error::Result;
-use cassandra_protocol::frame::Version;
-use rustls::ServerName;
 
 /// Single node TLS connection config.
 #[derive(Derivative, Clone)]
 #[derivative(Debug)]
 pub struct NodeRustlsConfig {
-    pub contact_points: Vec<SocketAddr>,
-    pub dns_name: ServerName,
+    pub(crate) contact_points: Vec<SocketAddr>,
+    pub(crate) dns_name: ServerName,
     #[derivative(Debug = "ignore")]
-    pub authenticator_provider: Arc<dyn SaslAuthenticatorProvider + Send + Sync>,
-    pub config: Arc<rustls::ClientConfig>,
-    pub version: Version,
-    pub beta_protocol: bool,
+    pub(crate) authenticator_provider: Arc<dyn SaslAuthenticatorProvider + Send + Sync>,
+    pub(crate) config: Arc<rustls::ClientConfig>,
+    pub(crate) version: Version,
+    pub(crate) beta_protocol: bool,
+    #[cfg(feature = "http-proxy")]
+    pub(crate) http_proxy: Option<HttpProxyConfig>,
 }
 
 /// Builder structure that helps to configure TLS connection for node.
@@ -32,6 +36,8 @@ pub struct NodeRustlsConfigBuilder {
     config: Arc<rustls::ClientConfig>,
     version: Version,
     beta_protocol: bool,
+    #[cfg(feature = "http-proxy")]
+    http_proxy: Option<HttpProxyConfig>,
 }
 
 impl NodeRustlsConfigBuilder {
@@ -43,6 +49,8 @@ impl NodeRustlsConfigBuilder {
             config,
             version: Version::V4,
             beta_protocol: false,
+            #[cfg(feature = "http-proxy")]
+            http_proxy: None,
         }
     }
 
@@ -85,6 +93,14 @@ impl NodeRustlsConfigBuilder {
         self
     }
 
+    /// Adds HTTP proxy configuration
+    #[cfg(feature = "http-proxy")]
+    #[must_use]
+    pub fn with_http_proxy(mut self, config: HttpProxyConfig) -> Self {
+        self.http_proxy = Some(config);
+        self
+    }
+
     /// Finalizes building process
     pub async fn build(self) -> Result<NodeRustlsConfig> {
         // replace with map() when async lambdas become available
@@ -100,6 +116,8 @@ impl NodeRustlsConfigBuilder {
             config: self.config,
             version: self.version,
             beta_protocol: self.beta_protocol,
+            #[cfg(feature = "http-proxy")]
+            http_proxy: self.http_proxy,
         })
     }
 }
diff --git a/cdrs-tokio/src/cluster/config_tcp.rs b/cdrs-tokio/src/cluster/config_tcp.rs
index 45c1ff7..4af9f33 100644
--- a/cdrs-tokio/src/cluster/config_tcp.rs
+++ b/cdrs-tokio/src/cluster/config_tcp.rs
@@ -5,17 +5,21 @@ use derivative::Derivative;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+#[cfg(feature = "http-proxy")]
+use crate::cluster::HttpProxyConfig;
 use crate::cluster::NodeAddress;
 
 /// Single node TCP connection config.
 #[derive(Derivative, Clone)]
 #[derivative(Debug)]
 pub struct NodeTcpConfig {
-    pub contact_points: Vec<SocketAddr>,
+    pub(crate) contact_points: Vec<SocketAddr>,
     #[derivative(Debug = "ignore")]
-    pub authenticator_provider: Arc<dyn SaslAuthenticatorProvider + Send + Sync>,
-    pub version: Version,
-    pub beta_protocol: bool,
+    pub(crate) authenticator_provider: Arc<dyn SaslAuthenticatorProvider + Send + Sync>,
+    pub(crate) version: Version,
+    pub(crate) beta_protocol: bool,
+    #[cfg(feature = "http-proxy")]
+    pub(crate) http_proxy: Option<HttpProxyConfig>,
 }
 
 /// Builder structure that helps to configure TCP connection for node.
@@ -27,6 +31,8 @@ pub struct NodeTcpConfigBuilder {
     authenticator_provider: Arc<dyn SaslAuthenticatorProvider + Send + Sync>,
     version: Version,
     beta_protocol: bool,
+    #[cfg(feature = "http-proxy")]
+    http_proxy: Option<HttpProxyConfig>,
 }
 
 impl Default for NodeTcpConfigBuilder {
@@ -36,6 +42,8 @@ impl Default for NodeTcpConfigBuilder {
             authenticator_provider: Arc::new(NoneAuthenticatorProvider),
             version: Version::V4,
             beta_protocol: false,
+            #[cfg(feature = "http-proxy")]
+            http_proxy: None,
         }
     }
 }
@@ -84,6 +92,14 @@ impl NodeTcpConfigBuilder {
         self
     }
 
+    /// Adds HTTP proxy configuration
+    #[cfg(feature = "http-proxy")]
+    #[must_use]
+    pub fn with_http_proxy(mut self, config: HttpProxyConfig) -> Self {
+        self.http_proxy = Some(config);
+        self
+    }
+
     /// Finalizes building process
     pub async fn build(self) -> Result<NodeTcpConfig> {
         // replace with map() when async lambdas become available
@@ -97,6 +113,8 @@ impl NodeTcpConfigBuilder {
             authenticator_provider: self.authenticator_provider,
             version: self.version,
             beta_protocol: self.beta_protocol,
+            #[cfg(feature = "http-proxy")]
+            http_proxy: self.http_proxy,
         })
     }
 }
diff --git a/cdrs-tokio/src/cluster/rustls_connection_manager.rs b/cdrs-tokio/src/cluster/rustls_connection_manager.rs
index 4a27cb1..e41ea73 100644
--- a/cdrs-tokio/src/cluster/rustls_connection_manager.rs
+++ b/cdrs-tokio/src/cluster/rustls_connection_manager.rs
@@ -1,11 +1,20 @@
+#[cfg(feature = "http-proxy")]
+use async_http_proxy::{http_connect_tokio, http_connect_tokio_with_basic_auth};
 use futures::FutureExt;
+use std::io;
+#[cfg(feature = "http-proxy")]
+use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::ops::Deref;
 use std::sync::Arc;
+#[cfg(feature = "http-proxy")]
+use tokio::net::TcpStream;
 use tokio::sync::mpsc::Sender;
 use tokio::time::sleep;
 
 use crate::cluster::connection_manager::{startup, ConnectionManager};
+#[cfg(feature = "http-proxy")]
+use crate::cluster::HttpProxyConfig;
 use crate::cluster::KeyspaceHolder;
 use crate::frame_encoding::FrameEncodingFactory;
 use crate::future::BoxFuture;
@@ -27,9 +36,12 @@ pub struct RustlsConnectionManager {
     buffer_size: usize,
     tcp_nodelay: bool,
     version: Version,
+    #[cfg(feature = "http-proxy")]
+    http_proxy: Option<HttpProxyConfig>,
 }
 
 impl ConnectionManager<TransportRustls> for RustlsConnectionManager {
+    //noinspection DuplicatedCode
     fn connection(
         &self,
         event_handler: Option<Sender<Envelope>>,
@@ -70,6 +82,7 @@ impl RustlsConnectionManager {
         buffer_size: usize,
         tcp_nodelay: bool,
         version: Version,
+        #[cfg(feature = "http-proxy")] http_proxy: Option<HttpProxyConfig>,
     ) -> Self {
         RustlsConnectionManager {
             dns_name,
@@ -82,16 +95,84 @@ impl RustlsConnectionManager {
             buffer_size,
             tcp_nodelay,
             version,
+            #[cfg(feature = "http-proxy")]
+            http_proxy,
         }
     }
 
-    async fn establish_connection(
+    //noinspection DuplicatedCode
+    #[cfg(feature = "http-proxy")]
+    async fn create_transport(
         &self,
         event_handler: Option<Sender<Envelope>>,
         error_handler: Option<Sender<Error>>,
         addr: SocketAddr,
-    ) -> Result<TransportRustls> {
-        let transport = TransportRustls::new(
+    ) -> io::Result<TransportRustls> {
+        if let Some(http_proxy) = &self.http_proxy {
+            let mut stream = TcpStream::connect(&http_proxy.address).await?;
+
+            if let Some(auth) = &http_proxy.basic_auth {
+                http_connect_tokio_with_basic_auth(
+                    &mut stream,
+                    &addr.ip().to_string(),
+                    addr.port(),
+                    &auth.username,
+                    &auth.password,
+                )
+                .await
+                .map_err(|error| io::Error::new(ErrorKind::Other, error.to_string()))?;
+            } else {
+                http_connect_tokio(&mut stream, &addr.ip().to_string(), addr.port())
+                    .await
+                    .map_err(|error| io::Error::new(ErrorKind::Other, error.to_string()))?;
+            }
+
+            stream.set_nodelay(self.tcp_nodelay)?;
+            TransportRustls::with_stream(
+                stream,
+                addr,
+                self.dns_name.clone(),
+                self.config.clone(),
+                self.keyspace_holder.clone(),
+                event_handler,
+                error_handler,
+                self.compression,
+                self.frame_encoder_factory
+                    .create_encoder(self.version, self.compression),
+                self.frame_encoder_factory
+                    .create_decoder(self.version, self.compression),
+                self.buffer_size,
+            )
+            .await
+        } else {
+            TransportRustls::new(
+                addr,
+                self.dns_name.clone(),
+                self.config.clone(),
+                self.keyspace_holder.clone(),
+                event_handler,
+                error_handler,
+                self.compression,
+                self.frame_encoder_factory
+                    .create_encoder(self.version, self.compression),
+                self.frame_encoder_factory
+                    .create_decoder(self.version, self.compression),
+                self.buffer_size,
+                self.tcp_nodelay,
+            )
+            .await
+        }
+    }
+
+    //noinspection DuplicatedCode
+    #[cfg(not(feature = "http-proxy"))]
+    async fn create_transport(
+        &self,
+        event_handler: Option<Sender<Envelope>>,
+        error_handler: Option<Sender<Error>>,
+        addr: SocketAddr,
+    ) -> io::Result<TransportRustls> {
+        TransportRustls::new(
             addr,
             self.dns_name.clone(),
             self.config.clone(),
@@ -106,7 +187,18 @@ impl RustlsConnectionManager {
             self.buffer_size,
             self.tcp_nodelay,
         )
-        .await?;
+        .await
+    }
+
+    async fn establish_connection(
+        &self,
+        event_handler: Option<Sender<Envelope>>,
+        error_handler: Option<Sender<Error>>,
+        addr: SocketAddr,
+    ) -> Result<TransportRustls> {
+        let transport = self
+            .create_transport(event_handler, error_handler, addr)
+            .await?;
 
         startup(
             &transport,
diff --git a/cdrs-tokio/src/cluster/session.rs b/cdrs-tokio/src/cluster/session.rs
index cef901e..152e52e 100644
--- a/cdrs-tokio/src/cluster/session.rs
+++ b/cdrs-tokio/src/cluster/session.rs
@@ -1195,6 +1195,8 @@ impl<LB: LoadBalancingStrategy<TransportTcp, TcpConnectionManager> + Send + Sync
                         self.config.transport_buffer_size,
                         self.config.tcp_nodelay,
                         self.node_config.version,
+                        #[cfg(feature = "http-proxy")]
+                        self.node_config.http_proxy,
                     );
 
                     self.config
@@ -1341,6 +1343,8 @@ impl<
                         self.config.transport_buffer_size,
                         self.config.tcp_nodelay,
                         self.node_config.version,
+                        #[cfg(feature = "http-proxy")]
+                        self.node_config.http_proxy,
                     );
 
                     self.config
diff --git a/cdrs-tokio/src/cluster/tcp_connection_manager.rs b/cdrs-tokio/src/cluster/tcp_connection_manager.rs
index 9088fb2..5176323 100644
--- a/cdrs-tokio/src/cluster/tcp_connection_manager.rs
+++ b/cdrs-tokio/src/cluster/tcp_connection_manager.rs
@@ -1,11 +1,20 @@
+#[cfg(feature = "http-proxy")]
+use async_http_proxy::{http_connect_tokio, http_connect_tokio_with_basic_auth};
 use futures::FutureExt;
+use std::io;
+#[cfg(feature = "http-proxy")]
+use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::ops::Deref;
 use std::sync::Arc;
+#[cfg(feature = "http-proxy")]
+use tokio::net::TcpStream;
 use tokio::sync::mpsc::Sender;
 use tokio::time::sleep;
 
 use crate::cluster::connection_manager::{startup, ConnectionManager};
+#[cfg(feature = "http-proxy")]
+use crate::cluster::HttpProxyConfig;
 use crate::cluster::KeyspaceHolder;
 use crate::frame_encoding::FrameEncodingFactory;
 use crate::future::BoxFuture;
@@ -25,9 +34,12 @@ pub struct TcpConnectionManager {
     buffer_size: usize,
     tcp_nodelay: bool,
     version: Version,
+    #[cfg(feature = "http-proxy")]
+    http_proxy: Option<HttpProxyConfig>,
 }
 
 impl ConnectionManager<TransportTcp> for TcpConnectionManager {
+    //noinspection DuplicatedCode
     fn connection(
         &self,
         event_handler: Option<Sender<Envelope>>,
@@ -65,6 +77,7 @@ impl TcpConnectionManager {
         buffer_size: usize,
         tcp_nodelay: bool,
         version: Version,
+        #[cfg(feature = "http-proxy")] http_proxy: Option<HttpProxyConfig>,
     ) -> Self {
         Self {
             authenticator_provider,
@@ -75,16 +88,79 @@ impl TcpConnectionManager {
             buffer_size,
             tcp_nodelay,
             version,
+            #[cfg(feature = "http-proxy")]
+            http_proxy,
         }
     }
 
-    async fn establish_connection(
+    //noinspection DuplicatedCode
+    #[cfg(feature = "http-proxy")]
+    async fn create_transport(
         &self,
         event_handler: Option<Sender<Envelope>>,
         error_handler: Option<Sender<Error>>,
         addr: SocketAddr,
-    ) -> Result<TransportTcp> {
-        let transport = TransportTcp::new(
+    ) -> io::Result<TransportTcp> {
+        if let Some(http_proxy) = &self.http_proxy {
+            let mut stream = TcpStream::connect(&http_proxy.address).await?;
+
+            if let Some(auth) = &http_proxy.basic_auth {
+                http_connect_tokio_with_basic_auth(
+                    &mut stream,
+                    &addr.ip().to_string(),
+                    addr.port(),
+                    &auth.username,
+                    &auth.password,
+                )
+                .await
+                .map_err(|error| io::Error::new(ErrorKind::Other, error.to_string()))?;
+            } else {
+                http_connect_tokio(&mut stream, &addr.ip().to_string(), addr.port())
+                    .await
+                    .map_err(|error| io::Error::new(ErrorKind::Other, error.to_string()))?;
+            }
+
+            stream.set_nodelay(self.tcp_nodelay)?;
+            TransportTcp::with_stream(
+                stream,
+                addr,
+                self.keyspace_holder.clone(),
+                event_handler,
+                error_handler,
+                self.compression,
+                self.frame_encoder_factory
+                    .create_encoder(self.version, self.compression),
+                self.frame_encoder_factory
+                    .create_decoder(self.version, self.compression),
+                self.buffer_size,
+            )
+        } else {
+            TransportTcp::new(
+                addr,
+                self.keyspace_holder.clone(),
+                event_handler,
+                error_handler,
+                self.compression,
+                self.frame_encoder_factory
+                    .create_encoder(self.version, self.compression),
+                self.frame_encoder_factory
+                    .create_decoder(self.version, self.compression),
+                self.buffer_size,
+                self.tcp_nodelay,
+            )
+            .await
+        }
+    }
+
+    //noinspection DuplicatedCode
+    #[cfg(not(feature = "http-proxy"))]
+    async fn create_transport(
+        &self,
+        event_handler: Option<Sender<Envelope>>,
+        error_handler: Option<Sender<Error>>,
+        addr: SocketAddr,
+    ) -> io::Result<TransportTcp> {
+        TransportTcp::new(
             addr,
             self.keyspace_holder.clone(),
             event_handler,
@@ -97,7 +173,18 @@ impl TcpConnectionManager {
             self.buffer_size,
             self.tcp_nodelay,
         )
-        .await?;
+        .await
+    }
+
+    async fn establish_connection(
+        &self,
+        event_handler: Option<Sender<Envelope>>,
+        error_handler: Option<Sender<Error>>,
+        addr: SocketAddr,
+    ) -> Result<TransportTcp> {
+        let transport = self
+            .create_transport(event_handler, error_handler, addr)
+            .await?;
 
         startup(
             &transport,
diff --git a/cdrs-tokio/src/transport.rs b/cdrs-tokio/src/transport.rs
index e8290c6..17f7138 100644
--- a/cdrs-tokio/src/transport.rs
+++ b/cdrs-tokio/src/transport.rs
@@ -103,22 +103,46 @@ impl TransportTcp {
     ) -> io::Result<TransportTcp> {
         TcpStream::connect(addr).await.and_then(move |socket| {
             socket.set_nodelay(tcp_nodelay)?;
+            Self::with_stream(
+                socket,
+                addr,
+                keyspace_holder,
+                event_handler,
+                error_handler,
+                compression,
+                frame_encoder,
+                frame_decoder,
+                buffer_size,
+            )
+        })
+    }
 
-            let (read_half, write_half) = split(socket);
-            Ok(TransportTcp {
-                inner: AsyncTransport::new(
-                    addr,
-                    compression,
-                    frame_encoder,
-                    frame_decoder,
-                    buffer_size,
-                    read_half,
-                    write_half,
-                    event_handler,
-                    error_handler,
-                    keyspace_holder,
-                ),
-            })
+    #[allow(clippy::too_many_arguments)]
+    pub fn with_stream<T: AsyncRead + AsyncWrite + Send + Sync + 'static>(
+        stream: T,
+        addr: SocketAddr,
+        keyspace_holder: Arc<KeyspaceHolder>,
+        event_handler: Option<mpsc::Sender<Envelope>>,
+        error_handler: Option<mpsc::Sender<Error>>,
+        compression: Compression,
+        frame_encoder: Box<dyn FrameEncoder + Send + Sync>,
+        frame_decoder: Box<dyn FrameDecoder + Send + Sync>,
+        buffer_size: usize,
+    ) -> io::Result<TransportTcp> {
+        let (read_half, write_half) = split(stream);
+        Ok(TransportTcp {
+            inner: AsyncTransport::new(
+                addr,
+                compression,
+                frame_encoder,
+                frame_decoder,
+                buffer_size,
+                read_half,
+                write_half,
+                event_handler,
+                error_handler,
+                keyspace_holder,
+            ),
         })
     }
 }
@@ -170,6 +194,36 @@ impl TransportRustls {
         let stream = TcpStream::connect(addr).await?;
         stream.set_nodelay(tcp_nodelay)?;
 
+        Self::with_stream(
+            stream,
+            addr,
+            dns_name,
+            config,
+            keyspace_holder,
+            event_handler,
+            error_handler,
+            compression,
+            frame_encoder,
+            frame_decoder,
+            buffer_size,
+        )
+        .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub async fn with_stream<T: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static>(
+        stream: T,
+        addr: SocketAddr,
+        dns_name: rustls::ServerName,
+        config: Arc<rustls::ClientConfig>,
+        keyspace_holder: Arc<KeyspaceHolder>,
+        event_handler: Option<mpsc::Sender<Envelope>>,
+        error_handler: Option<mpsc::Sender<Error>>,
+        compression: Compression,
+        frame_encoder: Box<dyn FrameEncoder + Send + Sync>,
+        frame_decoder: Box<dyn FrameDecoder + Send + Sync>,
+        buffer_size: usize,
+    ) -> io::Result<Self> {
         let connector = RustlsConnector::from(config.clone());
         let stream = connector.connect(dns_name, stream).await?;
         let (read_half, write_half) = split(stream);


Test patch: