*: bump proto

src/hstreamdb/src/client.rs

@@ -1,15 +1,17 @@
 use common::{Stream, Subscription};
 use hstreamdb_pb::h_stream_api_client::HStreamApiClient;
 use hstreamdb_pb::{
-    CompressionType, DeleteStreamRequest, DeleteSubscriptionRequest, ListStreamsRequest,
-    ListSubscriptionsRequest, LookupSubscriptionRequest, NodeState,
+    CompressionType, DeleteStreamRequest, DeleteSubscriptionRequest, GetStreamRequest,
+    GetSubscriptionRequest, ListConsumersRequest, ListStreamsRequest, ListSubscriptionsRequest,
+    LookupSubscriptionRequest, NodeState,
 };
 use tonic::transport::{Channel, ClientTlsConfig};
 use tonic::Request;
 use url::Url;
 
 use crate::appender::Appender;
 use crate::channel_provider::{new_channel_provider, ChannelProviderSettings, Channels};
+use crate::common::Error::PBUnwrapError;
 use crate::producer::{FlushCallback, FlushSettings, Producer};
 use crate::{common, flow_controller, format_url, producer};
 
@@ -135,6 +137,21 @@ impl Client {
             .streams;
         Ok(streams)
     }
+
+    pub async fn get_stream<T: Into<String>>(&self, stream_name: T) -> common::Result<Stream> {
+        let stream = self
+            .channels
+            .channel()
+            .await
+            .get_stream(GetStreamRequest {
+                name: stream_name.into(),
+            })
+            .await?
+            .into_inner()
+            .stream
+            .ok_or_else(|| PBUnwrapError("stream".into()))?;
+        Ok(stream)
+    }
 }
 
 impl Client {
@@ -182,6 +199,24 @@ impl Client {
             .collect();
         Ok(subscriptions)
     }
+
+    pub async fn get_subscription<T: Into<String>>(
+        &self,
+        subscription_id: T,
+    ) -> common::Result<Subscription> {
+        let subscription = self
+            .channels
+            .channel()
+            .await
+            .get_subscription(GetSubscriptionRequest {
+                id: subscription_id.into(),
+            })
+            .await?
+            .into_inner()
+            .subscription
+            .ok_or_else(|| PBUnwrapError("subscription".into()))?;
+        Ok(subscription.into())
+    }
 }
 
 impl Client {
@@ -248,6 +283,23 @@ impl Client {
     }
 }
 
+impl Client {
+    pub async fn list_consumers<T: Into<String>>(
+        &self,
+        subscription_id: T,
+    ) -> common::Result<Vec<common::Consumer>> {
+        let subscription_id: String = subscription_id.into();
+        Ok(self
+            .channels
+            .channel()
+            .await
+            .list_consumers(ListConsumersRequest { subscription_id })
+            .await?
+            .into_inner()
+            .consumers)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::env;

src/hstreamdb/src/common.rs

@@ -1,6 +1,6 @@
 use std::io;
 
-pub use hstreamdb_pb::{CompressionType, RecordId, SpecialOffset, Stream};
+pub use hstreamdb_pb::{CompressionType, Consumer, RecordId, SpecialOffset, Stream};
 use hstreamdb_pb::{HStreamRecord, StreamingFetchRequest};
 use num_bigint::ParseBigIntError;
 use prost::{DecodeError, Message};

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