fix API

src/connector/Cargo.toml

@@ -120,6 +120,7 @@ rumqttc = { version = "0.24.0", features = ["url"] }
 rust_decimal = "1"
 rustls-native-certs = "0.7"
 rustls-pemfile = "2"
+rustls-pki-types = "1"
 rw_futures_util = { workspace = true }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_derive = "1"
@@ -143,7 +144,6 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
 ] }
 tokio-postgres = { version = "0.7", features = ["with-uuid-1"] }
 tokio-retry = "0.3"
-tokio-rustls = "0.24"
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["codec", "io"] }
 tonic = { workspace = true }

src/connector/src/connector_common/common.rs

@@ -687,21 +687,21 @@ impl NatsCommon {
 
 pub(crate) fn load_certs(
     certificates: &str,
-) -> ConnectorResult<Vec<tokio_rustls::rustls::Certificate>> {
+) -> ConnectorResult<Vec<rustls_pki_types::CertificateDer<'static>>> {
     let cert_bytes = if let Some(path) = certificates.strip_prefix("fs://") {
         std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
     } else {
         certificates.as_bytes().to_owned()
     };
 
     rustls_pemfile::certs(&mut cert_bytes.as_slice())
-        .map(|cert| Ok(tokio_rustls::rustls::Certificate(cert?.to_vec())))
+        .map(|cert| Ok(cert?))
         .collect()
 }
 
 pub(crate) fn load_private_key(
     certificate: &str,
-) -> ConnectorResult<tokio_rustls::rustls::PrivateKey> {
+) -> ConnectorResult<rustls_pki_types::PrivateKeyDer<'static>> {
     let cert_bytes = if let Some(path) = certificate.strip_prefix("fs://") {
         std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
     } else {
@@ -711,7 +711,5 @@ pub(crate) fn load_private_key(
     let cert = rustls_pemfile::pkcs8_private_keys(&mut cert_bytes.as_slice())
         .next()
         .ok_or_else(|| anyhow!("No private key found"))?;
-    Ok(tokio_rustls::rustls::PrivateKey(
-        cert?.secret_pkcs8_der().to_vec(),
-    ))
+    Ok(cert?.into())
 }

src/connector/src/connector_common/mqtt_common.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use rumqttc::tokio_rustls::rustls;
 use rumqttc::v5::mqttbytes::QoS;
 use rumqttc::v5::{AsyncClient, EventLoop, MqttOptions};
 use serde_derive::Deserialize;
@@ -141,26 +142,22 @@ impl MqttCommon {
             .unwrap_or(QoS::AtMostOnce)
     }
 
-    fn get_tls_config(&self) -> ConnectorResult<tokio_rustls::rustls::ClientConfig> {
-        let mut root_cert_store = tokio_rustls::rustls::RootCertStore::empty();
+    fn get_tls_config(&self) -> ConnectorResult<rustls::ClientConfig> {
+        let mut root_cert_store = rustls::RootCertStore::empty();
         if let Some(ca) = &self.ca {
             let certificates = load_certs(ca)?;
             for cert in certificates {
-                root_cert_store.add(&cert).unwrap();
+                root_cert_store.add(cert).unwrap();
             }
         } else {
             for cert in
                 rustls_native_certs::load_native_certs().expect("could not load platform certs")
             {
-                root_cert_store
-                    .add(&tokio_rustls::rustls::Certificate(cert.to_vec()))
-                    .unwrap();
+                root_cert_store.add(cert).unwrap();
             }
         }
 
-        let builder = tokio_rustls::rustls::ClientConfig::builder()
-            .with_safe_defaults()
-            .with_root_certificates(root_cert_store);
+        let builder = rustls::ClientConfig::builder().with_root_certificates(root_cert_store);
 
         let tls_config = if let (Some(client_cert), Some(client_key)) =
             (self.client_cert.as_ref(), self.client_key.as_ref())

src/connector/src/error.rs

@@ -58,7 +58,7 @@ def_anyhow_newtype! {
     redis::RedisError => "Redis error",
     arrow_schema::ArrowError => "Arrow error",
     google_cloud_pubsub::client::google_cloud_auth::error::Error => "Google Cloud error",
-    tokio_rustls::rustls::Error => "TLS error",
+    rumqttc::tokio_rustls::rustls::Error => "TLS error",
     rumqttc::v5::ClientError => "MQTT error",
     rumqttc::v5::OptionError => "MQTT error",