shard reader: refactor

src/hstreamdb/src/client.rs

@@ -75,7 +75,7 @@ fn set_scheme(url: &str) -> Option<String> {
 }
 
 impl Client {
-    async fn new_channel_provider(
+    pub(crate) async fn new_channel_provider(
         &self,
         mut channel_provider_settings: ChannelProviderSettings,
     ) -> common::Result<Channels> {

src/hstreamdb/src/producer.rs

@@ -198,6 +198,7 @@ impl Producer {
                 }
             };
         }
+        log::debug!("producer channels closed, awaiting for all tasks to be finished");
 
         self.shard_buffer_timer
             .iter()
@@ -213,9 +214,11 @@ impl Producer {
         let tasks = std::mem::take(&mut self.tasks);
         for task in tasks {
             task.await.unwrap_or_else(|err| {
-                log::error!("await for task in stopping producer failed: {err}")
+                log::error!("failed to await for task when stopping producer: {err}")
             })
         }
+
+        log::info!("producer: graceful shutdown")
     }
 
     async fn handle_flush_request(&mut self, request: Option<ShardId>) {

src/hstreamdb/src/reader.rs

@@ -4,14 +4,17 @@ use hstreamdb_pb::{
 };
 use prost::DecodeError;
 
+use crate::channel_provider::Channels;
 use crate::client::Client;
 use crate::common::{self, ShardId};
 use crate::utils::decode_received_records;
-use crate::{format_url, Payload};
+use crate::{format_url, ChannelProviderSettings, Payload};
 
-pub struct ShardReaderId {
+pub struct ShardReader {
     reader_id: String,
     server_url: String,
+
+    channels: Channels,
 }
 
 impl Client {
@@ -22,7 +25,8 @@ impl Client {
         shard_id: ShardId,
         shard_offset: crate::common::StreamShardOffset,
         timeout_ms: u32,
-    ) -> common::Result<ShardReaderId> {
+        channel_provider_settings: ChannelProviderSettings,
+    ) -> common::Result<ShardReader> {
         let request = CreateShardReaderRequest {
             stream_name,
             shard_id,
@@ -49,24 +53,25 @@ impl Client {
             .ok_or_else(|| common::Error::PBUnwrapError("server_node".to_string()))?;
         let server_url = format_url!(&self.url_scheme, server_node);
 
-        Ok(ShardReaderId {
+        let channels = self.new_channel_provider(channel_provider_settings).await?;
+
+        Ok(ShardReader {
             reader_id,
             server_url,
+            channels,
         })
     }
+}
 
+impl ShardReader {
     pub async fn read_shard(
         &self,
-        shard_reader_id: &ShardReaderId,
         max_records: u32,
     ) -> common::Result<Vec<(RecordId, Result<Payload, DecodeError>)>> {
-        let mut channel = self
-            .channels
-            .channel_at(shard_reader_id.server_url.clone())
-            .await?;
+        let mut channel = self.channels.channel_at(self.server_url.clone()).await?;
         let records = channel
             .read_shard(ReadShardRequest {
-                reader_id: shard_reader_id.reader_id.clone(),
+                reader_id: self.reader_id.clone(),
                 max_records,
             })
             .await?
@@ -84,14 +89,11 @@ impl Client {
         Ok(records)
     }
 
-    pub async fn delete_shard_reader(&self, shard_reader_id: &ShardReaderId) -> common::Result<()> {
-        let mut channel = self
-            .channels
-            .channel_at(shard_reader_id.server_url.clone())
-            .await?;
+    pub async fn delete_shard_reader(self) -> common::Result<()> {
+        let mut channel = self.channels.channel_at(self.server_url).await?;
         channel
             .delete_shard_reader(DeleteShardReaderRequest {
-                reader_id: shard_reader_id.reader_id.clone(),
+                reader_id: self.reader_id,
             })
             .await?;
         Ok(())