erl: poc

Author: alissa-tung

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

@@ -1,20 +1,29 @@
 use hstreamdb::client::Client;
 use hstreamdb::producer::FlushSettings;
 use hstreamdb::{CompressionType, Record, Stream};
+use rustler::types::atom::ok;
 use rustler::{resource, Atom, Env, ResourceArc, Term};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 
 mod runtime;
 
 rustler::atoms! {
-    none, gzip, zstd
+    none, gzip, zstd,
+    len, size
 }
 
+rustler::init!(
+    "hstreamdb",
+    [create_stream, start_producer, append],
+    load = load
+);
+
 #[derive(Clone)]
 pub struct NifAppender(UnboundedSender<Record>);
 
 fn load(env: Env, _: Term) -> bool {
     resource!(NifAppender, env);
+    env_logger::init();
     true
 }
 
@@ -25,7 +34,7 @@ pub fn create_stream(
     replication_factor: u32,
     backlog_duration: u32,
     shard_count: u32,
-) {
+) -> Atom {
     let future = async move {
         let mut client = Client::new(url).await.unwrap();
         client
@@ -38,20 +47,21 @@ pub fn create_stream(
             .await
             .unwrap()
     };
-    _ = runtime::spawn(future)
+    _ = runtime::spawn(future);
+    ok()
 }
 
 #[rustler::nif]
 pub fn start_producer(
     url: String,
     stream_name: String,
     compression_type: Atom,
+    flush_settings: Term,
 ) -> ResourceArc<NifAppender> {
     let (request_sender, request_receiver) = unbounded_channel::<Record>();
+    let compression_type = atom_to_compression_type(compression_type);
+    let flush_settings = new_flush_settings(flush_settings);
     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)
@@ -72,16 +82,17 @@ pub fn start_producer(
 }
 
 #[rustler::nif]
-fn append(producer: ResourceArc<NifAppender>, partition_key: String, raw_payload: String) {
+fn append(producer: ResourceArc<NifAppender>, partition_key: String, raw_payload: String) -> Atom {
     let record = Record {
         partition_key,
         payload: hstreamdb::Payload::RawRecord(raw_payload.into_bytes()),
     };
     let producer = &producer.0;
     producer.send(record).unwrap();
+    ok()
 }
 
-pub fn atom_to_compression_type(compression_type: Atom) -> CompressionType {
+fn atom_to_compression_type(compression_type: Atom) -> CompressionType {
     if compression_type == none() {
         CompressionType::None
     } else if compression_type == gzip() {
@@ -93,8 +104,28 @@ pub fn atom_to_compression_type(compression_type: Atom) -> CompressionType {
     }
 }
 
-rustler::init!(
-    "hstreamdb",
-    [create_stream, start_producer, append],
-    load = load
-);
+fn new_flush_settings(proplists: Term) -> FlushSettings {
+    let proplists = proplists.into_list_iterator().unwrap();
+    let mut len_v = usize::MAX;
+    let mut size_v = usize::MAX;
+
+    for x in proplists {
+        if x.is_tuple() {
+            let (k, v): (Atom, usize) = x.decode().unwrap();
+            if k == len() {
+                len_v = v;
+            } else if k == size() {
+                size_v = v;
+            }
+        }
+    }
+
+    if len_v == usize::MAX && size_v == usize::MAX {
+        len_v = 0;
+        size_v = 0;
+    }
+    FlushSettings {
+        len: len_v,
+        size: size_v,
+    }
+}

x/hstreamdb_erl/.gitignore

@@ -0,0 +1,19 @@
+.rebar3
+_*
+.eunit
+*.o
+*.beam
+*.plt
+*.swp
+*.swo
+.erlang.cookie
+ebin
+log
+erl_crash.dump
+.rebar
+logs
+_build
+.idea
+*.iml
+rebar3.crashdump
+*~

x/hstreamdb_erl/rebar.config

@@ -0,0 +1,16 @@
+{erl_opts, [debug_info]}.
+
+{deps, []}.
+
+{project_plugins, [
+    erlfmt
+]}.
+
+{erlfmt, [write]}.
+
+{pre_hooks, [
+    {
+        compile,
+        "$(cd ../../ && cargo build --release)"
+    }
+]}.

x/hstreamdb_erl/rebar.lock

@@ -0,0 +1 @@
+[].

x/hstreamdb_erl/src/hstreamdb.erl

@@ -0,0 +1,46 @@
+-module(hstreamdb).
+
+-on_load(init/0).
+
+-export([create_stream/5, start_producer/4, append/3]).
+
+-export_type([producer/0, compression_type/0]).
+
+-type producer() :: any().
+-type compression_type() :: none | gzip | zstd.
+
+init() ->
+    Load = erlang:load_nif("../../target/release/libhstreamdb_erl_nifs", 0),
+    io:format("~p~n", [Load]).
+
+-spec create_stream(
+    ServerUrl :: binary(),
+    StreamName :: binary(),
+    ReplicationFactor :: pos_integer(),
+    BacklogDuration :: pos_integer(),
+    ShardCount :: pos_integer()
+) ->
+    ok.
+create_stream(
+    _ServerUrl,
+    _StreamName,
+    _ReplicationFactor,
+    _BacklogDuration,
+    _ShardCount
+) ->
+    none.
+
+-spec start_producer(
+    ServerUrl :: binary(),
+    StreamName :: binary(),
+    CompressionType :: compression_type(),
+    FlushSettings :: proplists:proplist()
+) ->
+    producer().
+start_producer(_ServerUrl, _StreamName, _CompressionType, _FlushSettings) ->
+    none.
+
+-spec append(Producer :: producer(), PartitionKey :: binary(), RawPayload :: binary()) ->
+    ok.
+append(_Producer, _PartitionKey, _RawPayload) ->
+    none.

x/hstreamdb_erl/src/hstreamdb_erl.app.src

@@ -0,0 +1,12 @@
+{application, hstreamdb_erl, [
+    {description, "An OTP library"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {applications, [
+        kernel,
+        stdlib
+    ]},
+    {env, []},
+    {modules, []},
+    {links, []}
+]}.

x/hstreamdb_erl/src/hstreamdb_erl.erl

@@ -0,0 +1,3 @@
+-module(hstreamdb_erl).
+
+-export([]).