Make TLS connection builders take references to configs instead of cl…

…oning (#1642)

shotover-proxy/tests/cassandra_int_tests/cluster/mod.rs

@@ -22,7 +22,7 @@ pub async fn run_topology_task(ca_path: Option<&str>, port: Option<u32>) -> Vec<
     let (keyspaces_tx, _keyspaces_rx) = watch::channel(HashMap::new());
     let (task_handshake_tx, task_handshake_rx) = mpsc::channel(1);
     let tls = ca_path.map(|ca_path| {
-        TlsConnector::new(TlsConnectorConfig {
+        TlsConnector::new(&TlsConnectorConfig {
             certificate_authority_path: ca_path.into(),
             certificate_path: None,
             private_key_path: None,

shotover/src/sources/cassandra.rs

@@ -75,7 +75,7 @@ impl CassandraSource {
             CassandraCodecBuilder::new(Direction::Source, name),
             Arc::new(Semaphore::new(connection_limit.unwrap_or(512))),
             trigger_shutdown_rx.clone(),
-            tls.map(TlsAcceptor::new).transpose()?,
+            tls.as_ref().map(TlsAcceptor::new).transpose()?,
             timeout.map(Duration::from_secs),
             transport.unwrap_or(Transport::Tcp),
         )

shotover/src/sources/kafka.rs

@@ -71,7 +71,7 @@ impl KafkaSource {
             KafkaCodecBuilder::new(Direction::Source, name),
             Arc::new(Semaphore::new(connection_limit.unwrap_or(512))),
             trigger_shutdown_rx.clone(),
-            tls.map(TlsAcceptor::new).transpose()?,
+            tls.as_ref().map(TlsAcceptor::new).transpose()?,
             timeout.map(Duration::from_secs),
             Transport::Tcp,
         )

shotover/src/sources/redis.rs

@@ -71,7 +71,7 @@ impl RedisSource {
             RedisCodecBuilder::new(Direction::Source, name),
             Arc::new(Semaphore::new(connection_limit.unwrap_or(512))),
             trigger_shutdown_rx.clone(),
-            tls.map(TlsAcceptor::new).transpose()?,
+            tls.as_ref().map(TlsAcceptor::new).transpose()?,
             timeout.map(Duration::from_secs),
             Transport::Tcp,
         )

shotover/src/tls.rs

@@ -74,14 +74,14 @@ fn load_private_key(path: &str) -> Result<PrivateKeyDer<'static>> {
 }
 
 impl TlsAcceptor {
-    pub fn new(tls_config: TlsAcceptorConfig) -> Result<TlsAcceptor, Vec<String>> {
+    pub fn new(tls_config: &TlsAcceptorConfig) -> Result<TlsAcceptor, Vec<String>> {
         // TODO: report multiple errors back to the user
         // The validation and anyhow error reporting has merged here and they were originally intended to be seperate.
         // We should probably replace the error reporting Vec<String> system with a better typed error system. (anyhow maybe?)
         Self::new_inner(tls_config).map_err(|x| vec![format!("{x:?}")])
     }
 
-    fn new_inner(tls_config: TlsAcceptorConfig) -> Result<TlsAcceptor> {
+    fn new_inner(tls_config: &TlsAcceptorConfig) -> Result<TlsAcceptor> {
         let client_cert_verifier =
             if let Some(path) = tls_config.certificate_authority_path.as_ref() {
                 let root_cert_store = load_ca(path).with_context(|| {
@@ -149,7 +149,7 @@ pub struct TlsConnector {
 }
 
 impl TlsConnector {
-    pub fn new(tls_config: TlsConnectorConfig) -> Result<TlsConnector> {
+    pub fn new(tls_config: &TlsConnectorConfig) -> Result<TlsConnector> {
         let root_cert_store =
             load_ca(&tls_config.certificate_authority_path).with_context(|| {
                 format!(

shotover/src/transforms/cassandra/sink_cluster/mod.rs

@@ -72,7 +72,7 @@ impl TransformConfig for CassandraSinkClusterConfig {
         &self,
         transform_context: TransformContextConfig,
     ) -> Result<Box<dyn TransformBuilder>> {
-        let tls = self.tls.clone().map(TlsConnector::new).transpose()?;
+        let tls = self.tls.as_ref().map(TlsConnector::new).transpose()?;
         let mut shotover_nodes = self.shotover_nodes.clone();
         let index = self
             .shotover_nodes

shotover/src/transforms/cassandra/sink_single.rs

@@ -37,7 +37,7 @@ impl TransformConfig for CassandraSinkSingleConfig {
         &self,
         transform_context: TransformContextConfig,
     ) -> Result<Box<dyn TransformBuilder>> {
-        let tls = self.tls.clone().map(TlsConnector::new).transpose()?;
+        let tls = self.tls.as_ref().map(TlsConnector::new).transpose()?;
         Ok(Box::new(CassandraSinkSingleBuilder::new(
             self.address.clone(),
             transform_context.chain_name,

shotover/src/transforms/kafka/sink_cluster/mod.rs

@@ -98,7 +98,7 @@ impl TransformConfig for KafkaSinkClusterConfig {
         &self,
         transform_context: TransformContextConfig,
     ) -> Result<Box<dyn TransformBuilder>> {
-        let tls = self.tls.clone().map(TlsConnector::new).transpose()?;
+        let tls = self.tls.as_ref().map(TlsConnector::new).transpose()?;
 
         let shotover_nodes: Result<Vec<_>> = self
             .shotover_nodes

shotover/src/transforms/kafka/sink_cluster/scram_over_mtls.rs

@@ -155,7 +155,7 @@ impl AuthorizeScramOverMtlsConfig {
         read_timeout: Option<Duration>,
     ) -> Result<AuthorizeScramOverMtlsBuilder> {
         let mtls_connection_factory = ConnectionFactory::new(
-            Some(TlsConnector::new(self.tls.clone())?),
+            Some(TlsConnector::new(&self.tls)?),
             connect_timeout,
             read_timeout,
             Arc::new(Notify::new()),

shotover/src/transforms/kafka/sink_single.rs

@@ -35,7 +35,7 @@ impl TransformConfig for KafkaSinkSingleConfig {
         &self,
         transform_context: TransformContextConfig,
     ) -> Result<Box<dyn TransformBuilder>> {
-        let tls = self.tls.clone().map(TlsConnector::new).transpose()?;
+        let tls = self.tls.as_ref().map(TlsConnector::new).transpose()?;
         Ok(Box::new(KafkaSinkSingleBuilder::new(
             self.destination_port,
             transform_context.chain_name,

shotover/src/transforms/redis/sink_single.rs

@@ -34,7 +34,7 @@ impl TransformConfig for RedisSinkSingleConfig {
         &self,
         transform_context: TransformContextConfig,
     ) -> Result<Box<dyn TransformBuilder>> {
-        let tls = self.tls.clone().map(TlsConnector::new).transpose()?;
+        let tls = self.tls.as_ref().map(TlsConnector::new).transpose()?;
         Ok(Box::new(RedisSinkSingleBuilder::new(
             self.address.clone(),
             tls,

shotover/src/transforms/util/cluster_connection_pool.rs

@@ -75,7 +75,7 @@ impl<C: CodecBuilder + 'static, A: Authenticator<T>, T: Token> ConnectionPool<C,
         Ok(Self {
             connect_timeout,
             lanes: Arc::new(Mutex::new(HashMap::new())),
-            tls: tls.map(TlsConnector::new).transpose()?,
+            tls: tls.as_ref().map(TlsConnector::new).transpose()?,
             codec,
             authenticator,
         })