x: try to export NIFs via rustler for Erlang client (#27)

* erl: init

* erl: wip

Author: alissa-tung

Cargo.lock

@@ -4,4 +4,5 @@ members = [
   "src/hstreamdb-pb",
   "src/utils/hstreamdb-test-utils",
   "src/utils/workspace-hack",
+  "src/x/hstreamdb-erl-nifs",
 ]

Cargo.toml

@@ -98,5 +98,17 @@ pub enum Payload {
     RawRecord(Vec<u8>),
 }
 
+impl From<Vec<u8>> for Payload {
+    fn from(payload: Vec<u8>) -> Self {
+        Self::RawRecord(payload)
+    }
+}
+
+impl From<prost_types::Struct> for Payload {
+    fn from(payload: prost_types::Struct) -> Self {
+        Self::HRecord(payload)
+    }
+}
+
 pub(crate) type ShardId = u64;
 pub type PartitionKey = String;

src/hstreamdb/src/common.rs

@@ -9,3 +9,4 @@ env_logger = "0.9.0"
 
 anyhow = "1.0.62"
 rand = "0.8.5"
+workspace-hack = { version = "0.1", path = "../workspace-hack" }

src/utils/hstreamdb-test-utils/Cargo.toml

@@ -20,8 +20,8 @@ log = { version = "0.4", default-features = false, features = ["std"] }
 prost = { version = "0.11", features = ["prost-derive", "std"] }
 prost-types = { version = "0.11", features = ["std"] }
 rand = { version = "0.8", features = ["alloc", "getrandom", "libc", "rand_chacha", "small_rng", "std", "std_rng"] }
-regex = { version = "1", default-features = false, features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode-bool"] }
-regex-syntax = { version = "0.6", default-features = false, features = ["unicode-bool"] }
+regex = { version = "1", features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
+regex-syntax = { version = "0.6", features = ["unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
 tokio = { version = "1", features = ["bytes", "io-std", "io-util", "libc", "macros", "memchr", "mio", "net", "num_cpus", "once_cell", "rt", "rt-multi-thread", "sync", "time", "tokio-macros"] }
 
 [build-dependencies]
@@ -31,8 +31,8 @@ indexmap = { version = "1", default-features = false, features = ["std"] }
 log = { version = "0.4", default-features = false, features = ["std"] }
 prost = { version = "0.11", features = ["prost-derive", "std"] }
 prost-types = { version = "0.11", features = ["std"] }
-regex = { version = "1", default-features = false, features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode-bool"] }
-regex-syntax = { version = "0.6", default-features = false, features = ["unicode-bool"] }
+regex = { version = "1", features = ["aho-corasick", "memchr", "perf", "perf-cache", "perf-dfa", "perf-inline", "perf-literal", "std", "unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
+regex-syntax = { version = "0.6", features = ["unicode", "unicode-age", "unicode-bool", "unicode-case", "unicode-gencat", "unicode-perl", "unicode-script", "unicode-segment"] }
 syn = { version = "1", features = ["clone-impls", "derive", "extra-traits", "full", "parsing", "printing", "proc-macro", "quote", "visit", "visit-mut"] }
 
 ### END HAKARI SECTION

src/utils/workspace-hack/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "hstreamdb-erl-nifs"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+workspace-hack = { version = "0.1", path = "../../utils/workspace-hack" }
+hstreamdb = { path = "../../hstreamdb" }
+
+log = "0.4.17"
+env_logger = "0.9.0"
+
+once_cell = "1.14.0"
+rustler = "0.26.0"
+tokio = { version = "1.21.0", features = ["rt-multi-thread"] }

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

@@ -0,0 +1,100 @@
+use hstreamdb::client::Client;
+use hstreamdb::producer::FlushSettings;
+use hstreamdb::{CompressionType, Record, Stream};
+use rustler::{resource, Atom, Env, ResourceArc, Term};
+use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
+
+mod runtime;
+
+rustler::atoms! {
+    none, gzip, zstd
+}
+
+#[derive(Clone)]
+pub struct NifAppender(UnboundedSender<Record>);
+
+fn load(env: Env, _: Term) -> bool {
+    resource!(NifAppender, env);
+    true
+}
+
+#[rustler::nif]
+pub fn create_stream(
+    url: String,
+    stream_name: String,
+    replication_factor: u32,
+    backlog_duration: u32,
+    shard_count: u32,
+) {
+    let future = async move {
+        let mut client = Client::new(url).await.unwrap();
+        client
+            .create_stream(Stream {
+                stream_name,
+                replication_factor,
+                backlog_duration,
+                shard_count,
+            })
+            .await
+            .unwrap()
+    };
+    _ = runtime::spawn(future)
+}
+
+#[rustler::nif]
+pub fn start_producer(
+    url: String,
+    stream_name: String,
+    compression_type: Atom,
+) -> ResourceArc<NifAppender> {
+    let (request_sender, request_receiver) = unbounded_channel::<Record>();
+    let future = async move {
+        let compression_type = atom_to_compression_type(compression_type);
+        let flush_settings = FlushSettings { len: 0, size: 0 };
+
+        let mut client = Client::new(url).await.unwrap();
+        let (appender, mut producer) = client
+            .new_producer(stream_name, compression_type, flush_settings)
+            .await
+            .unwrap();
+
+        _ = tokio::spawn(async move {
+            let mut request_receiver = request_receiver;
+            let mut appender = appender;
+            while let Some(record) = request_receiver.recv().await {
+                appender.append(record).unwrap()
+            }
+        });
+        producer.start().await
+    };
+    _ = runtime::spawn(future);
+    ResourceArc::new(NifAppender(request_sender))
+}
+
+#[rustler::nif]
+fn append(producer: ResourceArc<NifAppender>, partition_key: String, raw_payload: String) {
+    let record = Record {
+        partition_key,
+        payload: hstreamdb::Payload::RawRecord(raw_payload.into_bytes()),
+    };
+    let producer = &producer.0;
+    producer.send(record).unwrap();
+}
+
+pub fn atom_to_compression_type(compression_type: Atom) -> CompressionType {
+    if compression_type == none() {
+        CompressionType::None
+    } else if compression_type == gzip() {
+        CompressionType::Gzip
+    } else if compression_type == zstd() {
+        CompressionType::Zstd
+    } else {
+        panic!()
+    }
+}
+
+rustler::init!(
+    "hstreamdb",
+    [create_stream, start_producer, append],
+    load = load
+);

src/x/hstreamdb-erl-nifs/src/lib.rs

@@ -0,0 +1,24 @@
+use std::future::Future;
+
+use once_cell::sync::Lazy;
+use tokio::runtime::{self, Runtime};
+use tokio::task::JoinHandle;
+
+pub(crate) static TOKIO_RT: Lazy<Runtime> = Lazy::new(|| {
+    runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .map_err(|err| {
+            log::error!("failed to init Tokio runtime: {err}");
+            err
+        })
+        .unwrap()
+});
+
+pub(crate) fn spawn<T>(future: T) -> JoinHandle<T::Output>
+where
+    T: Future + Send + 'static,
+    T::Output: Send + 'static,
+{
+    TOKIO_RT.spawn(future)
+}