Merge pull request #887 from Migorithm/refactor/ch-type

rfct : ch type

Author: Migorithm

duva/src/domains/cluster_actors/queue.rs

@@ -156,10 +156,10 @@ impl ClusterActorSender {
 
     pub(crate) async fn route_subscribe_topology_change(
         &self,
-    ) -> anyhow::Result<tokio::sync::broadcast::Receiver<Topology>> {
+    ) -> tokio::sync::broadcast::Receiver<Topology> {
         let (tx, rx) = Callback::create();
-        self.send(ClientMessage::SubscribeToTopologyChange(tx)).await?;
-        Ok(rx.recv().await)
+        self.send(ClientMessage::SubscribeToTopologyChange(tx)).await;
+        rx.recv().await
     }
 }
 

duva/src/lib.rs

@@ -244,16 +244,14 @@ impl StartUpFacade {
             },
             ConnectionRequests::Authenticate(request) => {
                 let client_id = writer.send_conn_res(&self.cluster_actor_sender, request).await?;
-                let observer =
-                    self.cluster_actor_sender.route_subscribe_topology_change().await.unwrap();
-                let stream_writer = writer.run(observer);
+                let observer = self.cluster_actor_sender.route_subscribe_topology_change().await;
                 let client_controller = ClientController {
                     cluster_actor_sender: self.cluster_actor_sender.clone(),
                     cache_manager: self.cache_manager.clone(),
                 };
                 tokio::spawn(
                     ClientStreamReader { client_id, r: read_half }
-                        .handle_client_stream(client_controller, stream_writer),
+                        .handle_client_stream(client_controller, writer.run(observer)),
                 );
             },
         }

duva/src/presentation/clients/stream.rs

@@ -135,14 +135,14 @@ impl ClientStreamWriter {
 
         let (client_id, request_id) = auth_req.deconstruct()?;
 
-        self.serialized_write(ConnectionResponses::Authenticated(ConnectionResponse {
-            client_id: client_id.to_string(),
+        let connection_response = ConnectionResponse {
+            client_id: client_id.clone(),
             request_id,
             topology: cluster_manager.route_get_topology().await?,
             is_leader_node: replication_state.role == ReplicationRole::Leader,
-            replication_id: replication_state.replid.clone(),
-        }))
-        .await?;
+            replication_id: replication_state.replid,
+        };
+        self.serialized_write(ConnectionResponses::Authenticated(connection_response)).await?;
 
         Ok(client_id)
     }