re: RecordId

src/hstreamdb-pb/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 edition = "2021"
 name = "hstreamdb-pb"
-version = "0.1.0"
+version = "0.1.1"
 
 license = "BSD-3-Clause"
 description = "Rust protocol library for HStreamDB"

src/hstreamdb-pb/src/lib.rs

@@ -1,2 +1,14 @@
 #![allow(clippy::derive_partial_eq_without_eq)]
 tonic::include_proto!("hstream.server");
+
+use std::fmt;
+
+impl fmt::Display for RecordId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "{}-{}-{}",
+            self.shard_id, self.batch_id, self.batch_index
+        )
+    }
+}

src/hstreamdb/Cargo.toml

@@ -21,7 +21,7 @@ num-bigint = "0.4.3"
 num-traits = "0.2.15"
 md-5 = "0.10.1"
 
-hstreamdb-pb = { version = "0.1.0", path = "../hstreamdb-pb/" }
+hstreamdb-pb = { version = "0.1.1", path = "../hstreamdb-pb/" }
 prost-types = "0.11.1"
 
 prost = "0.11.0"

src/hstreamdb/src/appender.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use hstreamdb_pb::RecordId;
 use tokio::sync::oneshot;
 
 use crate::common::{self, Record};
@@ -28,7 +29,7 @@ impl Appender {
     pub async fn append(
         &mut self,
         record: Record,
-    ) -> common::Result<oneshot::Receiver<Result<String, Arc<common::Error>>>> {
+    ) -> common::Result<oneshot::Receiver<Result<RecordId, Arc<common::Error>>>> {
         if let Some(flow_controller) = &self.flow_controller {
             flow_controller.acquire(record.encoded_len()).await?
         }

src/hstreamdb/src/common.rs

@@ -1,7 +1,7 @@
 use std::io;
 
 use hstreamdb_pb::StreamingFetchRequest;
-pub use hstreamdb_pb::{CompressionType, SpecialOffset, Stream};
+pub use hstreamdb_pb::{CompressionType, RecordId, SpecialOffset, Stream};
 use num_bigint::ParseBigIntError;
 use prost::Message;
 use tonic::transport;

src/hstreamdb/src/consumer.rs

@@ -84,7 +84,7 @@ async fn fetching(
     }
 }
 
-type AckFn = Box<dyn FnOnce() -> Result<(), SendError<StreamingFetchRequest>> + Send>;
+pub type AckFn = Box<dyn FnOnce() -> Result<(), SendError<StreamingFetchRequest>> + Send>;
 
 fn process_streaming_fetch_response(
     consumer_name: String,

src/hstreamdb/src/lib.rs

@@ -11,5 +11,5 @@ pub mod utils;
 
 pub use channel_provider::ChannelProviderSettings;
 pub use common::{
-    CompressionType, Error, Payload, Record, Result, SpecialOffset, Stream, Subscription,
+    CompressionType, Error, Payload, Record, RecordId, Result, SpecialOffset, Stream, Subscription,
 };

src/hstreamdb/src/producer.rs

@@ -12,7 +12,7 @@ use hstreamdb_pb::h_stream_api_client::HStreamApiClient;
 use hstreamdb_pb::h_stream_record_header::Flag;
 use hstreamdb_pb::{
     AppendRequest, BatchHStreamRecords, BatchedRecord, CompressionType, HStreamRecord,
-    HStreamRecordHeader, ListShardsRequest, Shard,
+    HStreamRecordHeader, ListShardsRequest, RecordId, Shard,
 };
 use prost::Message;
 use tokio::select;
@@ -25,12 +25,12 @@ use crate::common::{self, PartitionKey, Record, ShardId};
 use crate::flow_controller::FlowControllerClient;
 use crate::utils::{self, partition_key_to_shard_id};
 
-type ResultVec = Vec<oneshot::Sender<Result<String, Arc<common::Error>>>>;
+type ResultVec = Vec<oneshot::Sender<Result<RecordId, Arc<common::Error>>>>;
 
 #[derive(Debug)]
 pub(crate) struct Request(
     pub(crate) Record,
-    pub(crate) oneshot::Sender<Result<String, Arc<common::Error>>>,
+    pub(crate) oneshot::Sender<Result<RecordId, Arc<common::Error>>>,
 );
 
 pub struct Producer {
@@ -405,7 +405,7 @@ async fn append(
     shard_id: ShardId,
     compression_type: CompressionType,
     records: Vec<Record>,
-) -> common::Result<Vec<String>> {
+) -> common::Result<Vec<RecordId>> {
     let (batch_size, payload) = batch_records(compression_type, records)?;
     let records = BatchedRecord {
         compression_type: compression_type as i32,
@@ -424,9 +424,7 @@ async fn append(
         .await?
         .into_inner()
         .record_ids
-        .iter()
-        .map(utils::record_id_to_string)
-        .collect::<Vec<_>>();
+        .to_vec();
     Ok(record_ids)
 }
 

src/hstreamdb/src/utils.rs

@@ -1,5 +1,5 @@
 use hstreamdb_pb::h_stream_api_client::HStreamApiClient;
-use hstreamdb_pb::{LookupShardRequest, RecordId, Shard};
+use hstreamdb_pb::{LookupShardRequest, Shard};
 use md5::{Digest, Md5};
 use num_bigint::BigInt;
 use num_traits::Num;
@@ -8,13 +8,6 @@ use tonic::transport::Channel;
 use crate::common::{self, PartitionKey, ShardId};
 use crate::{format_url, Error};
 
-pub fn record_id_to_string(record_id: &RecordId) -> String {
-    format!(
-        "{}-{}-{}",
-        record_id.shard_id, record_id.batch_id, record_id.batch_index
-    )
-}
-
 pub async fn lookup_shard(
     channel: &mut HStreamApiClient<Channel>,
     url_scheme: &str,

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

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use hstreamdb::client::Client;
 use hstreamdb::producer::FlushSettings;
-use hstreamdb::{ChannelProviderSettings, CompressionType, Record, Stream};
+use hstreamdb::{ChannelProviderSettings, CompressionType, Record, RecordId, Stream};
 use once_cell::sync::OnceCell;
 use parking_lot::{Mutex, MutexGuard};
 use rustler::types::atom::{error, ok};
@@ -39,7 +39,7 @@ enum AppendResult {
 
 struct AppendResultFuture(Mutex<Option<AppendResultType>>, OnceCell<AppendResult>);
 
-type AppendResultType = oneshot::Receiver<Result<String, Arc<hstreamdb::Error>>>;
+type AppendResultType = oneshot::Receiver<Result<RecordId, Arc<hstreamdb::Error>>>;
 #[derive(Clone)]
 pub struct NifAppender(UnboundedSender<Option<(Record, oneshot::Sender<AppendResultType>)>>);
 
@@ -207,7 +207,11 @@ fn await_append_result(env: Env, x: ResourceArc<AppendResultFuture>) -> Term {
         let receiver: &Mutex<_> = &x.0;
         let mut receiver: MutexGuard<Option<_>> = receiver.lock();
         let receiver = mem::take(&mut (*receiver));
-        let append_result: Result<String, Arc<_>> = receiver.unwrap().blocking_recv().unwrap();
+        let append_result: Result<String, Arc<_>> = receiver
+            .unwrap()
+            .blocking_recv()
+            .unwrap()
+            .map(|x| x.to_string());
         let append_result = match append_result {
             Ok(record_id) => RecordId(record_id),
             Err(err) => Error(err.to_string()),