erl: tls (#82)

Author: alissa-tung

Cargo.lock

@@ -13,7 +13,7 @@ keywords = ["hstreamdb", "streaming-database", "database-client"]
 [dependencies]
 prost = "0.11.0"
 prost-types = "0.11.1"
-tonic = "0.8.0"
+tonic = "0.8.2"
 workspace-hack = { version = "0.1", path = "../utils/workspace-hack" }
 
 [build-dependencies]

src/hstreamdb-pb/Cargo.toml

@@ -95,9 +95,9 @@ impl ChannelProviderSettingsBuilder {
         })
     }
 
-    pub fn set_client_tls_config(self, client_tls_config: ClientTlsConfig) -> Self {
+    pub fn set_tls_config(self, tls_config: ClientTlsConfig) -> Self {
         Self(ChannelProviderSettings {
-            client_tls_config: Some(client_tls_config),
+            client_tls_config: Some(tls_config),
             ..self.0
         })
     }

src/hstreamdb/src/channel_provider.rs

@@ -4,7 +4,7 @@ use hstreamdb_pb::{
     CompressionType, DeleteStreamRequest, DeleteSubscriptionRequest, ListStreamsRequest,
     ListSubscriptionsRequest, LookupSubscriptionRequest, NodeState,
 };
-use tonic::transport::Channel;
+use tonic::transport::{Channel, ClientTlsConfig};
 use tonic::Request;
 use url::Url;
 
@@ -16,6 +16,7 @@ use crate::{common, flow_controller, format_url, producer};
 pub struct Client {
     pub(crate) channels: Channels,
     pub(crate) url_scheme: String,
+    tls_config: Option<ClientTlsConfig>,
 }
 
 impl Client {
@@ -46,6 +47,7 @@ impl Client {
             }
         };
         let mut hstream_api_client = HStreamApiClient::connect(String::from(url)).await?;
+        let tls_config = channel_provider_settings.client_tls_config.clone();
         let channels = new_channel_provider(
             &url_scheme,
             &mut hstream_api_client,
@@ -55,15 +57,19 @@ impl Client {
         Ok(Client {
             channels,
             url_scheme,
+            tls_config,
         })
     }
 }
 
 impl Client {
     async fn new_channel_provider(
         &self,
-        channel_provider_settings: ChannelProviderSettings,
+        mut channel_provider_settings: ChannelProviderSettings,
     ) -> common::Result<Channels> {
+        if channel_provider_settings.client_tls_config.is_none() {
+            channel_provider_settings.client_tls_config = self.tls_config.clone()
+        }
         new_channel_provider(
             &self.url_scheme,
             &mut self.channels.channel().await,

src/hstreamdb/src/client.rs

@@ -20,3 +20,4 @@ tokio = { version = "1.21.0", features = ["rt-multi-thread", "parking_lot"] }
 tokio-stream = "0.1.11"
 
 prost = "0.11.0"
+tonic = "0.8.2"

src/x/hstreamdb-erl-nifs/Cargo.toml

@@ -18,6 +18,7 @@ use rustler::{
 };
 use tokio::sync::{oneshot, Mutex, MutexGuard};
 use tokio_stream::StreamExt;
+use tonic::transport::{Certificate, ClientTlsConfig, Identity};
 
 mod runtime;
 
@@ -40,6 +41,7 @@ rustler::atoms! {
     eos, already_acked,
     create_shard_reader_reply, read_shard_reply,
     bad_hstream_record,
+    tls_config,
 }
 
 rustler::init!(
@@ -56,6 +58,10 @@ rustler::init!(
         async_ack,
         async_create_shard_reader,
         async_read_shard,
+        new_client_tls_config,
+        set_domain_name,
+        set_ca_certificate,
+        set_identity,
     ],
     load = load
 );
@@ -145,25 +151,61 @@ fn load(env: Env, _: Term) -> bool {
     resource!(AppendResultFuture, env);
     resource!(NifResponder, env);
     resource!(NifShardReaderId, env);
+    resource!(NifClientTlsConfig, env);
     env_logger::init();
     true
 }
 
 #[rustler::nif]
-fn async_start_client(pid: LocalPid, url: String, _conf: Term) {
-    let future = async move {
-        let client = Client::new(url, ChannelProviderSettings::builder().build()).await;
-        OwnedEnv::new().send_and_clear(&pid, |env| match client {
-            Ok(client) => (
-                start_client_reply(),
-                ok(),
-                ResourceArc::new(NifClient(client)),
-            )
-                .encode(env),
-            Err(err) => (start_client_reply(), error(), err.to_string()).encode(env),
-        })
-    };
-    runtime::spawn(future);
+fn async_start_client<'a>(env: Env<'a>, pid: LocalPid, url: String, options: Term) -> Term<'a> {
+    match from_start_client_options(options) {
+        Err(err) => (error(), (badarg(), err.to_string())).encode(env),
+        Ok(channel_provider_settings) => {
+            let future = async move {
+                let client = Client::new(url, channel_provider_settings).await;
+                OwnedEnv::new().send_and_clear(&pid, |env| match client {
+                    Ok(client) => (
+                        start_client_reply(),
+                        ok(),
+                        ResourceArc::new(NifClient(client)),
+                    )
+                        .encode(env),
+                    Err(err) => (start_client_reply(), error(), err.to_string()).encode(env),
+                })
+            };
+            runtime::spawn(future);
+            ok().to_term(env)
+        }
+    }
+}
+
+fn from_start_client_options(proplists: Term) -> hstreamdb::Result<ChannelProviderSettings> {
+    let proplists = proplists
+        .into_list_iterator()
+        .map_err(|err| hstreamdb::Error::BadArgument(format!("{err:?}")))?;
+
+    let mut channel_provider_settings = ChannelProviderSettings::builder();
+    for x in proplists {
+        if x.is_tuple() {
+            let (k, v): (Atom, Term) = x
+                .decode()
+                .map_err(|err| hstreamdb::Error::BadArgument(format!("{err:?}")))?;
+            if k == tls_config() {
+                let v: ResourceArc<NifClientTlsConfig> = v
+                    .decode()
+                    .map_err(|err| hstreamdb::Error::BadArgument(format!("{err:?}")))?;
+                let NifClientTlsConfig(v) = (*v).clone();
+                channel_provider_settings = channel_provider_settings.set_tls_config(v)
+            } else if k == concurrency_limit() {
+                let v: usize = v
+                    .decode()
+                    .map_err(|err| hstreamdb::Error::BadArgument(format!("{err:?}")))?;
+                channel_provider_settings = channel_provider_settings.set_concurrency_limit(v)
+            }
+        }
+    }
+
+    todo!()
 }
 
 #[rustler::nif]
@@ -352,16 +394,14 @@ fn async_stop_producer(pid: LocalPid, producer: ResourceArc<NifAppender>) {
     runtime::spawn(future);
 }
 
-fn try_append<'a>(
-    env: Env<'a>,
+#[rustler::nif]
+fn async_append(
     pid: LocalPid,
     producer: ResourceArc<NifAppender>,
     partition_key: String,
-    raw_payload: Term,
-) -> Result<(), Term<'a>> {
-    let raw_payload = rustler::Binary::from_term(raw_payload)
-        .map_err(|err| (badarg(), format!("{err:?}")).encode(env))?
-        .to_vec();
+    raw_payload: Binary,
+) {
+    let raw_payload = raw_payload.to_vec();
     let future = async move {
         let record = Record {
             partition_key,
@@ -376,21 +416,6 @@ fn try_append<'a>(
         }
     };
     runtime::spawn(future);
-    Ok(())
-}
-
-#[rustler::nif]
-fn async_append<'a>(
-    env: Env<'a>,
-    pid: LocalPid,
-    producer: ResourceArc<NifAppender>,
-    partition_key: String,
-    raw_payload: Term,
-) -> Term<'a> {
-    match try_append(env, pid, producer, partition_key, raw_payload) {
-        Ok(()) => ok().encode(env),
-        Err(err) => (error(), err).encode(env),
-    }
 }
 
 #[rustler::nif]
@@ -732,3 +757,44 @@ fn async_read_shard(
     };
     runtime::spawn(future);
 }
+
+#[derive(Clone)]
+struct NifClientTlsConfig(ClientTlsConfig);
+
+#[rustler::nif]
+fn new_client_tls_config() -> ResourceArc<NifClientTlsConfig> {
+    ResourceArc::new(NifClientTlsConfig(ClientTlsConfig::new()))
+}
+
+#[rustler::nif]
+fn set_domain_name(
+    tls_config: ResourceArc<NifClientTlsConfig>,
+    domain_name: String,
+) -> ResourceArc<NifClientTlsConfig> {
+    let NifClientTlsConfig(tls_config) = (*tls_config).clone();
+    ResourceArc::new(NifClientTlsConfig(tls_config.domain_name(domain_name)))
+}
+
+#[rustler::nif]
+fn set_ca_certificate(
+    tls_config: ResourceArc<NifClientTlsConfig>,
+    ca_certificate: Binary,
+) -> ResourceArc<NifClientTlsConfig> {
+    let NifClientTlsConfig(tls_config) = (*tls_config).clone();
+    ResourceArc::new(NifClientTlsConfig(
+        tls_config.ca_certificate(Certificate::from_pem(ca_certificate.as_slice())),
+    ))
+}
+
+#[rustler::nif]
+fn set_identity(
+    tls_config: ResourceArc<NifClientTlsConfig>,
+    cert: Binary,
+    key: Binary,
+) -> ResourceArc<NifClientTlsConfig> {
+    let NifClientTlsConfig(tls_config) = &*tls_config;
+    let tls_config = tls_config.clone();
+    ResourceArc::new(NifClientTlsConfig(
+        tls_config.identity(Identity::from_pem(cert.as_slice(), key.as_slice())),
+    ))
+}