erl: append return result value, for bad arg and closed producer (#69)

src/hstreamdb/src/common.rs

@@ -58,9 +58,9 @@ pub enum Error {
     StreamingFetchInitError(tokio::sync::mpsc::error::SendError<StreamingFetchRequest>),
     #[error("failed to unwrap `{0}`")]
     PBUnwrapError(String),
-    #[error("No channel is available")]
+    #[error("no channel is available")]
     NoChannelAvailable,
-    #[error("{0}")]
+    #[error("bad argument: {0}")]
     BadArgument(String),
     #[error(transparent)]
     AppenderSendError(producer::SendError),

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

@@ -6,13 +6,14 @@ use hstreamdb::producer::{FlushCallback, FlushSettings};
 use hstreamdb::{ChannelProviderSettings, CompressionType, Record, RecordId, Stream};
 use once_cell::sync::OnceCell;
 use parking_lot::{Mutex, MutexGuard};
-use rustler::types::atom::{error, ok};
+use rustler::types::atom::{badarg, error, ok};
 use rustler::{resource, Atom, Encoder, Env, LocalPid, NifResult, OwnedEnv, ResourceArc, Term};
 use tokio::sync::oneshot;
 
 mod runtime;
 
 rustler::atoms! {
+    terminated,
     compression_type, none, gzip, zstd,
     concurrency_limit, flow_control_size,
     max_batch_len, max_batch_size, batch_deadline,
@@ -187,24 +188,41 @@ fn stop_producer(producer: ResourceArc<NifAppender>) -> Atom {
     ok()
 }
 
-#[rustler::nif]
-fn append(
+fn try_append<'a>(
+    env: Env<'a>,
     producer: ResourceArc<NifAppender>,
     partition_key: String,
-    raw_payload: String,
-) -> ResourceArc<AppendResultFuture> {
+    raw_payload: Term,
+) -> Result<ResourceArc<AppendResultFuture>, Term<'a>> {
+    let raw_payload = rustler::Binary::from_term(raw_payload)
+        .map_err(|err| (badarg(), format!("{err:?}")).encode(env))?;
     let record = Record {
         partition_key,
-        payload: hstreamdb::Payload::RawRecord(raw_payload.into_bytes()),
+        payload: hstreamdb::Payload::RawRecord(raw_payload.to_vec()),
     };
     let producer = &producer.0;
     let (sender, receiver) = oneshot::channel();
-    producer.send(Some((record, sender))).unwrap();
+    producer
+        .send(Some((record, sender)))
+        .map_err(|_| terminated().encode(env))?;
     let receiver = receiver.blocking_recv().unwrap();
-    ResourceArc::new(AppendResultFuture(
+    Ok(ResourceArc::new(AppendResultFuture(
         Mutex::new(Some(receiver)),
         OnceCell::new(),
-    ))
+    )))
+}
+
+#[rustler::nif]
+fn append<'a>(
+    env: Env<'a>,
+    producer: ResourceArc<NifAppender>,
+    partition_key: String,
+    raw_payload: Term,
+) -> Term<'a> {
+    match try_append(env, producer, partition_key, raw_payload) {
+        Ok(x) => (ok(), x).encode(env),
+        Err(err) => (error(), err).encode(env),
+    }
 }
 
 #[rustler::nif]