feat: add get_roles

Author: echo724

duva-client/src/controller.rs

@@ -48,7 +48,6 @@ impl<T> ClientController<T> {
             | Config { .. }
             | Info
             | ClusterForget { .. }
-            | Role
             | ReplicaOf { .. }
             | ClusterInfo => match query_io {
                 | QueryIO::Null => Response::Null,
@@ -130,7 +129,7 @@ impl<T> ClientController<T> {
                 }
                 Response::Array(keys)
             },
-            | ClusterNodes => {
+            | Role | ClusterNodes => {
                 let QueryIO::Array(value) = query_io else {
                     return Response::FormatError;
                 };

duva/src/domains/cluster_actors/command.rs

@@ -62,6 +62,7 @@ pub enum ClientMessage {
     LeaderReqConsensus(ConsensusRequest),
     ClusterNodes(Callback<Vec<PeerState>>),
     GetRole(Callback<ReplicationRole>),
+    GetRoles(Callback<Vec<(PeerIdentifier, ReplicationRole)>>),
     SubscribeToTopologyChange(Callback<tokio::sync::broadcast::Receiver<Topology>>),
     ClusterMeet(PeerIdentifier, LazyOption, Callback<anyhow::Result<()>>),
     GetTopology(Callback<Topology>),

duva/src/domains/cluster_actors/service.rs

@@ -6,6 +6,7 @@ use crate::domains::cluster_actors::ConnectionMessage;
 use crate::domains::cluster_actors::SchedulerMessage;
 use crate::domains::operation_logs::interfaces::TWriteAheadLog;
 use crate::domains::peers::PeerMessage;
+use std::iter;
 
 use crate::prelude::PeerIdentifier;
 use crate::res_err;
@@ -109,6 +110,20 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
             | GetRole(callback) => {
                 callback.send(self.replication.role.clone());
             },
+            | GetRoles(callback) => {
+                let followers = self
+                    .members
+                    .iter()
+                    .filter(|(_, peer_state)| peer_state.is_replica(&self.replication.replid))
+                    .map(|(peer_id, peer_state)| (peer_id.clone(), peer_state.role().clone()));
+
+                let roles =
+                    iter::once((self.replication.self_identifier(), self.replication.role.clone()))
+                        .chain(followers)
+                        .collect();
+
+                callback.send(roles);
+            },
             | SubscribeToTopologyChange(callback) => {
                 callback.send(self.node_change_broadcast.subscribe());
             },

duva/src/presentation/clients/controller.rs

@@ -143,8 +143,15 @@ impl ClientController {
                 QueryIO::SimpleString("OK".into())
             },
             | ClientAction::Role => {
-                let role = self.cluster_communication_manager.route_get_role();
-                QueryIO::SimpleString(role.await?.to_string().into())
+                let roles = self.cluster_communication_manager.route_get_roles().await?;
+                QueryIO::Array(
+                    roles
+                        .into_iter()
+                        .map(|(peer_id, role)| {
+                            QueryIO::BulkString(format!("{}:{}", peer_id, role).into())
+                        })
+                        .collect(),
+                )
             },
             | ClientAction::Ttl { key } => {
                 QueryIO::SimpleString(self.cache_manager.route_ttl(key).await?.into())

duva/src/presentation/clusters/communication_manager.rs

@@ -106,6 +106,14 @@ impl ClusterCommunicationManager {
         Ok(rx.await?)
     }
 
+    pub(crate) async fn route_get_roles(
+        &self,
+    ) -> anyhow::Result<Vec<(PeerIdentifier, ReplicationRole)>> {
+        let (tx, rx) = Callback::create();
+        self.send(ClientMessage::GetRoles(tx)).await?;
+        Ok(rx.await?)
+    }
+
     pub(crate) async fn route_get_role(&self) -> anyhow::Result<ReplicationRole> {
         let (tx, rx) = Callback::create();
         self.send(ClientMessage::GetRole(tx)).await?;

duva/tests/common.rs

@@ -135,7 +135,8 @@ pub fn spawn_server_process(env: &ServerEnv) -> anyhow::Result<TestProcessChild>
                 continue;
             }
 
-            if role_res.is_empty() || (role_res != "leader" && role_res != "follower") {
+            if role_res.is_empty() || (role_res.contains("leader") && role_res.contains("follower"))
+            {
                 continue;
             }