*: bump proto

src/hstreamdb-pb/Cargo.toml

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

src/hstreamdb/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 edition = "2021"
 name = "hstreamdb"
-version = "0.1.0"
+version = "0.2.0"
 
 license = "BSD-3-Clause"
 description = "Rust client library for HStreamDB"
@@ -21,7 +21,7 @@ num-bigint = "0.4.3"
 num-traits = "0.2.15"
 md-5 = "0.10.1"
 
-hstreamdb-pb = { version = "0.1.1", path = "../hstreamdb-pb/" }
+hstreamdb-pb = { version = "0.2.0", path = "../hstreamdb-pb/" }
 prost-types = "0.11.1"
 
 prost = "0.11.0"

src/hstreamdb/src/client.rs

@@ -100,9 +100,9 @@ pub(crate) async fn get_available_node_addrs(
 }
 
 impl Client {
-    pub async fn create_stream(&self, stream: Stream) -> common::Result<()> {
-        self.channels.channel().await.create_stream(stream).await?;
-        Ok(())
+    pub async fn create_stream(&self, stream: Stream) -> common::Result<Stream> {
+        let stream = self.channels.channel().await.create_stream(stream).await?;
+        Ok(stream.into_inner())
     }
 
     pub async fn delete_stream(
@@ -138,14 +138,18 @@ impl Client {
 }
 
 impl Client {
-    pub async fn create_subscription(&self, subscription: Subscription) -> common::Result<()> {
+    pub async fn create_subscription(
+        &self,
+        subscription: Subscription,
+    ) -> common::Result<Subscription> {
         let subscription: hstreamdb_pb::Subscription = subscription.into();
-        self.channels
+        let subscription = self
+            .channels
             .channel()
             .await
             .create_subscription(subscription)
             .await?;
-        Ok(())
+        Ok(subscription.into_inner().into())
     }
 
     pub async fn delete_subscription(
@@ -267,12 +271,13 @@ mod tests {
             replication_factor: 1,
             backlog_duration: 30 * 60,
             shard_count: 1,
+            creation_time: None,
         };
-        let streams = (0..10)
+        let mut streams = (0..10)
             .map(|_| make_stream(format!("stream-{}", rand_alphanumeric(10))))
             .collect::<Vec<_>>();
-        for stream in streams.iter() {
-            client.create_stream(stream.clone()).await.unwrap()
+        for stream in streams.iter_mut() {
+            *stream = client.create_stream(stream.clone()).await.unwrap();
         }
 
         let listed_streams = client.list_streams().await.unwrap();
@@ -302,12 +307,13 @@ mod tests {
             replication_factor: 1,
             backlog_duration: 30 * 60,
             shard_count: 1,
+            creation_time: None,
         };
-        let streams = (0..10)
+        let mut streams = (0..10)
             .map(|_| make_stream(format!("stream-{}", rand_alphanumeric(10))))
             .collect::<Vec<_>>();
-        for stream in streams.iter() {
-            client.create_stream(stream.clone()).await.unwrap()
+        for stream in streams.iter_mut() {
+            *stream = client.create_stream(stream.clone()).await.unwrap();
         }
 
         let listed_streams = client.list_streams().await.unwrap();
@@ -321,6 +327,7 @@ mod tests {
             ack_timeout_seconds: 60 * 10,
             max_unacked_records: 1000,
             offset: SpecialOffset::Earliest,
+            creation_time: None,
         };
         for stream in streams.iter() {
             let subscription_ids = (0..5)
@@ -333,7 +340,7 @@ mod tests {
                         stream.stream_name.clone(),
                     ))
                     .await
-                    .unwrap()
+                    .unwrap();
             }
             for subscription_id in subscription_ids.iter() {
                 assert!(

src/hstreamdb/src/common.rs

@@ -4,7 +4,7 @@ pub use hstreamdb_pb::{CompressionType, RecordId, SpecialOffset, Stream};
 use hstreamdb_pb::{HStreamRecord, StreamingFetchRequest};
 use num_bigint::ParseBigIntError;
 use prost::{DecodeError, Message};
-use prost_types::Struct;
+pub use prost_types::{ListValue, Struct, Timestamp};
 use tonic::transport;
 
 use crate::producer;
@@ -16,6 +16,7 @@ pub struct Subscription {
     pub ack_timeout_seconds: i32,
     pub max_unacked_records: i32,
     pub offset: SpecialOffset,
+    pub creation_time: Option<Timestamp>,
 }
 
 #[derive(Debug, Clone)]
@@ -48,6 +49,7 @@ impl From<Subscription> for hstreamdb_pb::Subscription {
             ack_timeout_seconds: subscription.ack_timeout_seconds,
             max_unacked_records: subscription.max_unacked_records,
             offset: subscription.offset as _,
+            creation_time: subscription.creation_time,
         }
     }
 }
@@ -61,6 +63,7 @@ impl From<hstreamdb_pb::Subscription> for Subscription {
             ack_timeout_seconds: subscription.ack_timeout_seconds,
             max_unacked_records: subscription.max_unacked_records,
             offset,
+            creation_time: subscription.creation_time,
         }
     }
 }

src/hstreamdb/src/flow_controller.rs

@@ -128,6 +128,7 @@ mod tests {
                 replication_factor: 1,
                 backlog_duration: 10 * 60,
                 shard_count: 40,
+                creation_time: None,
             })
             .await
             .unwrap();

src/hstreamdb/src/lib.rs

@@ -13,6 +13,6 @@ 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,
+    CompressionType, Error, ListValue, Payload, Record, RecordId, Result, ShardId, SpecialOffset,
+    Stream, StreamShardOffset, Struct, Subscription, Timestamp,
 };

src/hstreamdb/src/utils.rs

@@ -145,6 +145,7 @@ mod tests {
                 replication_factor: 1,
                 backlog_duration: 10 * 60,
                 shard_count: 200,
+                creation_time: None,
             })
             .await
             .unwrap();

src/hstreamdb/tests/consumer_test.rs

@@ -31,6 +31,7 @@ async fn test_consumer() {
             replication_factor: 1,
             backlog_duration: 30 * 60,
             shard_count: 12,
+            creation_time: None,
         })
         .await
         .unwrap();
@@ -42,6 +43,7 @@ async fn test_consumer() {
             ack_timeout_seconds: 60 * 60,
             max_unacked_records: 1000,
             offset: SpecialOffset::Earliest as _,
+            creation_time: None,
         })
         .await
         .unwrap();

src/hstreamdb/tests/producer_test.rs

@@ -28,6 +28,7 @@ async fn test_producer() {
             replication_factor: 1,
             backlog_duration: 30 * 60,
             shard_count: 12,
+            creation_time: None,
         })
         .await
         .unwrap();

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

@@ -225,6 +225,7 @@ fn async_create_stream(
                     replication_factor,
                     backlog_duration,
                     shard_count,
+                    creation_time: None,
                 })
                 .await?;
             Ok::<(), hstreamdb::Error>(())
@@ -262,10 +263,11 @@ fn async_create_subscription(
                         ack_timeout_seconds,
                         max_unacked_records,
                         offset,
+                        creation_time: None,
                     })
                     .await;
                 OwnedEnv::new().send_and_clear(&pid, |env| match result {
-                    Ok(()) => (create_subscription_reply(), ok()).encode(env),
+                    Ok(_) => (create_subscription_reply(), ok()).encode(env),
                     Err(err) => (create_subscription_reply(), error(), err.to_string()).encode(env),
                 })
             };