erl: refine client APIs

src/hstreamdb/src/consumer.rs

@@ -13,10 +13,10 @@ use crate::utils::decode_received_records;
 
 impl Client {
     pub async fn streaming_fetch(
-        &mut self,
+        &self,
         consumer_name: String,
         subscription_id: String,
-    ) -> common::Result<UnboundedReceiverStream<(Payload, AckFn)>> {
+    ) -> common::Result<UnboundedReceiverStream<(Payload, Responder)>> {
         let channel = self.lookup_subscription(subscription_id.clone()).await?;
         let mut channel = HStreamApiClient::connect(channel).await?;
 
@@ -36,7 +36,7 @@ impl Client {
             .send(request)
             .map_err(common::Error::StreamingFetchInitError)?;
 
-        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<(Payload, AckFn)>();
+        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<(Payload, Responder)>();
 
         _ = tokio::spawn(fetching(
             consumer_name,
@@ -55,7 +55,7 @@ async fn fetching(
     subscription_id: String,
     ack_sender: UnboundedSender<StreamingFetchRequest>,
     mut init_response: Streaming<StreamingFetchResponse>,
-    fetch_stream: UnboundedSender<(Payload, AckFn)>,
+    fetch_stream: UnboundedSender<(Payload, Responder)>,
 ) {
     loop {
         match init_response.message().await {
@@ -79,14 +79,26 @@ async fn fetching(
     }
 }
 
-pub type AckFn = Box<dyn FnOnce() -> Result<(), SendError<StreamingFetchRequest>> + Send>;
+type AckFn = Box<dyn FnOnce() -> Result<(), SendError<StreamingFetchRequest>> + Send>;
+
+pub struct Responder(AckFn);
+
+impl Responder {
+    pub fn ack(self) -> Result<(), SendError<StreamingFetchRequest>> {
+        self.0()
+    }
+
+    fn new(ack_fn: AckFn) -> Self {
+        Responder(ack_fn)
+    }
+}
 
 fn process_streaming_fetch_response(
     consumer_name: String,
     subscription_id: String,
     message: StreamingFetchResponse,
     ack_sender: UnboundedSender<StreamingFetchRequest>,
-    fetch_stream: UnboundedSender<(Payload, AckFn)>,
+    fetch_stream: UnboundedSender<(Payload, Responder)>,
 ) {
     match message.received_records {
         None => {
@@ -130,7 +142,7 @@ fn process_streaming_fetch_response(
                             ack_ids: vec![record_id],
                         })
                     });
-                    match fetch_stream.send((record, ack_fn)) {
+                    match fetch_stream.send((record, Responder::new(ack_fn))) {
                         Ok(()) => (),
                         Err(err) => {
                             log::error!("send to fetch stream error: {err}")

src/hstreamdb/src/lib.rs

@@ -11,6 +11,7 @@ pub mod reader;
 pub mod utils;
 
 pub use channel_provider::ChannelProviderSettings;
+pub use client::Client;
 pub use common::{
     CompressionType, Error, Payload, Record, RecordId, Result, ShardId, SpecialOffset, Stream,
     StreamShardOffset, Subscription,

src/hstreamdb/tests/consumer_test.rs

@@ -13,7 +13,7 @@ async fn test_consumer() {
     env_logger::init();
 
     let addr = env::var("TEST_SERVER_ADDR").unwrap();
-    let mut client = Client::new(
+    let client = Client::new(
         addr,
         ChannelProviderSettings::builder()
             .set_concurrency_limit(8)
@@ -105,10 +105,10 @@ async fn test_consumer() {
         .unwrap();
 
     let mut records = Vec::new();
-    while let Some((record, ack)) = stream.next().await {
+    while let Some((record, responder)) = stream.next().await {
         println!("{record:?}");
         records.push(record);
-        ack().unwrap();
+        responder.ack().unwrap();
         if records.len() == 10 * 100 {
             println!("done");
             break;

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

@@ -15,6 +15,8 @@ env_logger = "0.9.0"
 
 flume = "0.10.14"
 once_cell = "1.14.0"
-parking_lot = "0.12.1"
 rustler = "0.26.0"
 tokio = { version = "1.21.0", features = ["rt-multi-thread", "parking_lot"] }
+tokio-stream = "0.1.11"
+
+prost = "0.11.0"