pb: exported pb types should use enum directly (#19)

* pb: pb: exported pb types should use enum directly

* fix check

Author: alissa-tung

src/hstreamdb/src/client.rs

@@ -1,8 +1,8 @@
-use common::Stream;
+use common::{Stream, Subscription};
 use hstreamdb_pb::h_stream_api_client::HStreamApiClient;
 use hstreamdb_pb::{
     CompressionType, DeleteStreamRequest, DeleteSubscriptionRequest, ListStreamsRequest,
-    ListSubscriptionsRequest, LookupSubscriptionRequest, NodeState, Subscription,
+    ListSubscriptionsRequest, LookupSubscriptionRequest, NodeState,
 };
 use tonic::transport::Channel;
 use tonic::Request;
@@ -98,6 +98,7 @@ impl Client {
 
 impl Client {
     pub async fn create_subscription(&mut self, subscription: Subscription) -> common::Result<()> {
+        let subscription: hstreamdb_pb::Subscription = subscription.into();
         self.hstream_api_client
             .create_subscription(subscription)
             .await?;
@@ -126,7 +127,10 @@ impl Client {
             .list_subscriptions(ListSubscriptionsRequest {})
             .await?
             .into_inner()
-            .subscription;
+            .subscription
+            .into_iter()
+            .map(|x| x.into())
+            .collect();
         Ok(subscriptions)
     }
 }
@@ -183,10 +187,11 @@ impl Client {
 mod tests {
     use std::env;
 
-    use hstreamdb_pb::{SpecialOffset, Stream, Subscription};
+    use hstreamdb_pb::{SpecialOffset, Stream};
     use hstreamdb_test_utils::rand_alphanumeric;
 
     use super::Client;
+    use crate::Subscription;
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_stream_cld() {
@@ -249,7 +254,7 @@ mod tests {
             stream_name,
             ack_timeout_seconds: 60 * 10,
             max_unacked_records: 1000,
-            offset: SpecialOffset::Earliest as i32,
+            offset: SpecialOffset::Earliest,
         };
         for stream in streams.iter() {
             let subscription_ids = (0..5)

src/hstreamdb/src/common.rs

@@ -1,10 +1,43 @@
 use std::io;
 
 use hstreamdb_pb::StreamingFetchRequest;
-pub use hstreamdb_pb::{Stream, Subscription};
+pub use hstreamdb_pb::{SpecialOffset, Stream};
 use num_bigint::ParseBigIntError;
 use tonic::transport;
 
+pub struct Subscription {
+    pub subscription_id: String,
+    pub stream_name: String,
+    pub ack_timeout_seconds: i32,
+    pub max_unacked_records: i32,
+    pub offset: SpecialOffset,
+}
+
+impl From<Subscription> for hstreamdb_pb::Subscription {
+    fn from(subscription: Subscription) -> Self {
+        hstreamdb_pb::Subscription {
+            subscription_id: subscription.subscription_id,
+            stream_name: subscription.stream_name,
+            ack_timeout_seconds: subscription.ack_timeout_seconds,
+            max_unacked_records: subscription.max_unacked_records,
+            offset: subscription.offset as _,
+        }
+    }
+}
+
+impl From<hstreamdb_pb::Subscription> for Subscription {
+    fn from(subscription: hstreamdb_pb::Subscription) -> Self {
+        let offset = subscription.offset();
+        Subscription {
+            subscription_id: subscription.subscription_id,
+            stream_name: subscription.stream_name,
+            ack_timeout_seconds: subscription.ack_timeout_seconds,
+            max_unacked_records: subscription.max_unacked_records,
+            offset,
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum Error {
     TransportError(transport::Error),

src/hstreamdb/src/producer.rs

@@ -57,7 +57,7 @@ impl BufferState {
     }
 
     fn check(&self, flush_settings: &FlushSettings) -> bool {
-        (flush_settings.len >= self.len) || (flush_settings.size >= self.size)
+        (self.len >= flush_settings.len) || (self.size >= flush_settings.size)
     }
 }
 
@@ -157,27 +157,33 @@ async fn flush(
     compression_type: CompressionType,
     buffer: Vec<Record>,
 ) -> Result<(), String> {
-    match lookup_shard(&mut channel, &url_scheme, shard_id).await {
-        Err(err) => {
-            log::warn!("{err}");
-            Ok(())
-        }
-        Ok(server_node) => {
-            let channel = HStreamApiClient::connect(server_node.clone())
+    if !buffer.is_empty() {
+        match lookup_shard(&mut channel, &url_scheme, shard_id).await {
+            Err(err) => {
+                log::warn!("{err}");
+                Ok(())
+            }
+            Ok(server_node) => {
+                let channel = HStreamApiClient::connect(server_node.clone())
+                    .await
+                    .map_err(|err| {
+                        format!("producer connect error: addr = {server_node}, {err}")
+                    })?;
+                append(
+                    channel,
+                    stream_name,
+                    shard_id,
+                    compression_type,
+                    buffer.to_vec(),
+                )
                 .await
-                .map_err(|err| format!("producer connect error: addr = {server_node}, {err}"))?;
-            append(
-                channel,
-                stream_name,
-                shard_id,
-                compression_type,
-                buffer.to_vec(),
-            )
-            .await
-            .map_err(|err| format!("producer append error: addr = {server_node}, {err:?}"))
-            .map(|x| log::debug!("append succeed: len = {}", x.len()))?;
-            Ok(())
+                .map_err(|err| format!("producer append error: addr = {server_node}, {err:?}"))
+                .map(|x| log::debug!("append succeed: len = {}", x.len()))?;
+                Ok(())
+            }
         }
+    } else {
+        Ok(())
     }
 }