Repository: robustmq/robustmq (language: rust)

Issue:
RBIP-19: MQTT Command supports viewing connection information
<img width="1318" alt="image" src="https://github.com/user-attachments/assets/2d417789-e787-45a1-84c7-0dd700728b1c">


Discussion / hints:
Please assign me. Thanks.

Fix patch:
diff --git a/src/cli-command/src/mqtt.rs b/src/cli-command/src/mqtt.rs
index 97f6dcdf4..d1181d11b 100644
--- a/src/cli-command/src/mqtt.rs
+++ b/src/cli-command/src/mqtt.rs
@@ -15,12 +15,14 @@
 use std::sync::Arc;
 
 use grpc_clients::mqtt::admin::call::{
-    cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user, mqtt_broker_list_user,
+    cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user, mqtt_broker_list_connection,
+    mqtt_broker_list_user,
 };
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::user::MqttUser;
 use protocol::broker_mqtt::broker_mqtt_admin::{
-    ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, ListUserRequest,
+    ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, ListConnectionRequest,
+    ListUserRequest,
 };
 
 use crate::{error_info, grpc_addr};
@@ -34,9 +36,14 @@ pub struct MqttCliCommandParam {
 #[derive(Clone)]
 pub enum MqttActionType {
     Status,
+
+    // User admin
     CreateUser(CreateUserRequest),
     DeleteUser(DeleteUserRequest),
     ListUser,
+
+    // connection
+    ListConnection,
 }
 
 pub struct MqttBrokerCommand {}
@@ -69,6 +76,10 @@ impl MqttBrokerCommand {
             MqttActionType::ListUser => {
                 self.list_user(client_pool.clone(), params.clone()).await;
             }
+            MqttActionType::ListConnection => {
+                self.list_connections(client_pool.clone(), params.clone())
+                    .await;
+            }
         }
     }
 
@@ -144,4 +155,22 @@ impl MqttBrokerCommand {
             }
         }
     }
+
+    async fn list_connections(&self, client_pool: Arc<ClientPool>, params: MqttCliCommandParam) {
+        let request = ListConnectionRequest {};
+        match mqtt_broker_list_connection(client_pool.clone(), grpc_addr(params.server), request)
+            .await
+        {
+            Ok(data) => {
+                println!("connection list:");
+                for raw in data.list_connection_raw {
+                    println!("{:?}", raw)
+                }
+            }
+            Err(e) => {
+                println!("MQTT broker list connection exception");
+                error_info(e.to_string());
+            }
+        }
+    }
 }
diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml
index 507a2eddc..1ba9bfe94 100644
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -17,7 +17,7 @@ name = "cmd"
 version.workspace = true
 edition.workspace = true
 license.workspace = true
-default-run = "journal-server"
+default-run = "cli-command"
 
 [[bin]]
 name = "mqtt-server"
diff --git a/src/cmd/src/cli-command/command.rs b/src/cmd/src/cli-command/command.rs
index 7c6378fc3..91672edfa 100644
--- a/src/cmd/src/cli-command/command.rs
+++ b/src/cmd/src/cli-command/command.rs
@@ -57,9 +57,13 @@ struct MqttArgs {
 #[derive(Debug, Subcommand)]
 enum MQTTAction {
     Status,
+    // User admin
     CreateUser(CreateUserArgs),
     DeleteUser(DeleteUserArgs),
     ListUser,
+
+    // Connections
+    ListConnection,
 }
 
 #[derive(clap::Args, Debug)]
@@ -157,6 +161,7 @@ async fn main() {
                         username: arg.username,
                     }),
                     MQTTAction::ListUser => MqttActionType::ListUser,
+                    MQTTAction::ListConnection => MqttActionType::ListConnection,
                 },
             };
             cmd.start(params).await;
diff --git a/src/common/base/src/tools.rs b/src/common/base/src/tools.rs
index c4023b6d1..b61982e64 100644
--- a/src/common/base/src/tools.rs
+++ b/src/common/base/src/tools.rs
@@ -18,6 +18,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use local_ip_address::local_ip;
 use log::warn;
+use tonic::Status;
 use uuid::Uuid;
 
 use crate::error::common::CommonError;
@@ -133,6 +134,23 @@ pub fn read_file(path: &str) -> Result<String, CommonError> {
     Ok(fs::read_to_string(path)?)
 }
 
+/// Serialize a DashMap object into a JSON string
+/// # parameters
+/// * ` value ` - a reference to a DashMap object that implements the Serialize trait
+///
+/// # Return value
+/// Returns a result type that includes a string containing the serialized JSON data upon success and a Status object containing an error message upon failure
+///
+/// # Error
+/// If the serialization fails, return a Status object containing an error message
+pub fn serialize_value<T>(value: &T) -> Result<String, Status>
+where
+    T: serde::Serialize,
+{
+    serde_json::to_string(value)
+        .map_err(|e| Status::cancelled(CommonError::CommonError(e.to_string()).to_string()))
+}
+
 #[cfg(test)]
 mod tests {
     use crate::tools::{get_local_ip, unique_id};
diff --git a/src/common/metadata-struct/src/mqtt/connection.rs b/src/common/metadata-struct/src/mqtt/connection.rs
index 994fc1b46..f3796c88b 100644
--- a/src/common/metadata-struct/src/mqtt/connection.rs
+++ b/src/common/metadata-struct/src/mqtt/connection.rs
@@ -44,8 +44,10 @@ pub struct MQTTConnection {
     // Flags whether to return a detailed error message to the client when an error occurs.
     pub request_problem_info: u8,
     // Flow control part keeps track of how many QOS 1 and QOS 2 messages are still pending on the connection
+    #[serde(skip_serializing, skip_deserializing)]
     pub receive_qos_message: Arc<AtomicIsize>,
     // Flow control part keeps track of how many QOS 1 and QOS 2 messages are still pending on the connection
+    #[serde(skip_serializing, skip_deserializing)]
     pub sender_qos_message: Arc<AtomicIsize>,
     // Time when the connection was created
     pub create_time: u64,
diff --git a/src/grpc-clients/src/mqtt/admin/call.rs b/src/grpc-clients/src/mqtt/admin/call.rs
index 4c9c6037c..01445da05 100644
--- a/src/grpc-clients/src/mqtt/admin/call.rs
+++ b/src/grpc-clients/src/mqtt/admin/call.rs
@@ -18,12 +18,13 @@ use common_base::error::common::CommonError;
 use prost::Message as _;
 use protocol::broker_mqtt::broker_mqtt_admin::{
     ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
-    DeleteUserRequest, ListUserReply, ListUserRequest,
+    DeleteUserRequest, ListConnectionReply, ListConnectionRequest, ListUserReply, ListUserRequest,
 };
 
 use crate::mqtt::{retry_call, MqttBrokerPlacementInterface, MqttBrokerService};
 use crate::pool::ClientPool;
 
+// ---- cluster ------
 pub async fn cluster_status(
     client_pool: Arc<ClientPool>,
     addrs: Vec<String>,
@@ -47,6 +48,7 @@ pub async fn cluster_status(
     }
 }
 
+// ------ user -------
 pub async fn mqtt_broker_list_user(
     client_pool: Arc<ClientPool>,
     addrs: Vec<String>,
@@ -115,3 +117,27 @@ pub async fn mqtt_broker_delete_user(
         Err(e) => Err(e),
     }
 }
+
+// ------- connection  -----------
+pub async fn mqtt_broker_list_connection(
+    client_pool: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: ListConnectionRequest,
+) -> Result<ListConnectionReply, CommonError> {
+    let request_date = ListConnectionRequest::encode_to_vec(&request);
+    match retry_call(
+        MqttBrokerService::Admin,
+        MqttBrokerPlacementInterface::ListConnection,
+        client_pool,
+        addrs,
+        request_date,
+    )
+    .await
+    {
+        Ok(data) => match ListConnectionReply::decode(data.as_ref()) {
+            Ok(data) => Ok(data),
+            Err(e) => Err(CommonError::CommonError(e.to_string())),
+        },
+        Err(e) => Err(e),
+    }
+}
diff --git a/src/grpc-clients/src/mqtt/admin/inner.rs b/src/grpc-clients/src/mqtt/admin/inner.rs
index 361df01e3..4463d3afb 100644
--- a/src/grpc-clients/src/mqtt/admin/inner.rs
+++ b/src/grpc-clients/src/mqtt/admin/inner.rs
@@ -17,11 +17,12 @@ use mobc::Connection;
 use prost::Message;
 use protocol::broker_mqtt::broker_mqtt_admin::{
     ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
-    DeleteUserRequest, ListUserReply, ListUserRequest,
+    DeleteUserRequest, ListConnectionReply, ListConnectionRequest, ListUserReply, ListUserRequest,
 };
 
 use super::MqttBrokerAdminServiceManager;
 
+// ---- cluster -------
 pub(crate) async fn inner_cluster_status(
     mut client: Connection<MqttBrokerAdminServiceManager>,
     request: Vec<u8>,
@@ -35,6 +36,7 @@ pub(crate) async fn inner_cluster_status(
     }
 }
 
+// ----- user -------
 pub(crate) async fn inner_list_user(
     mut client: Connection<MqttBrokerAdminServiceManager>,
     request: Vec<u8>,
@@ -73,3 +75,17 @@ pub(crate) async fn inner_delete_user(
         Err(e) => Err(CommonError::CommonError(e.to_string())),
     }
 }
+
+// ------- connection ------
+pub(crate) async fn inner_list_connection(
+    mut client: Connection<MqttBrokerAdminServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, CommonError> {
+    match ListConnectionRequest::decode(request.as_ref()) {
+        Ok(request) => match client.mqtt_broker_list_connection(request).await {
+            Ok(result) => Ok(ListConnectionReply::encode_to_vec(&result.into_inner())),
+            Err(e) => Err(CommonError::GrpcServerStatus(e)),
+        },
+        Err(e) => Err(CommonError::CommonError(e.to_string())),
+    }
+}
diff --git a/src/grpc-clients/src/mqtt/admin/mod.rs b/src/grpc-clients/src/mqtt/admin/mod.rs
index 126cb32cd..efdbad348 100644
--- a/src/grpc-clients/src/mqtt/admin/mod.rs
+++ b/src/grpc-clients/src/mqtt/admin/mod.rs
@@ -21,6 +21,7 @@ use protocol::broker_mqtt::broker_mqtt_admin::mqtt_broker_admin_service_client::
 use tonic::transport::Channel;
 
 use super::MqttBrokerPlacementInterface;
+use crate::mqtt::admin::inner::inner_list_connection;
 use crate::pool::ClientPool;
 
 pub mod call;
@@ -55,6 +56,9 @@ pub(crate) async fn admin_interface_call(
                 MqttBrokerPlacementInterface::DeleteUser => {
                     inner_delete_user(client, request).await
                 }
+                MqttBrokerPlacementInterface::ListConnection => {
+                    inner_list_connection(client, request).await
+                }
                 _ => {
                     return Err(CommonError::CommonError(format!(
                         "admin service does not support service interfaces [{:?}]",
diff --git a/src/grpc-clients/src/mqtt/mod.rs b/src/grpc-clients/src/mqtt/mod.rs
index 526875db8..29bb71563 100644
--- a/src/grpc-clients/src/mqtt/mod.rs
+++ b/src/grpc-clients/src/mqtt/mod.rs
@@ -37,11 +37,14 @@ pub enum MqttBrokerPlacementInterface {
     UpdateCache,
     SendLastWillMessage,
 
-    // admin
+    // admin service  functions
     ClusterStatus,
     ListUser,
     CreateUser,
     DeleteUser,
+
+    // connection
+    ListConnection,
 }
 
 pub mod admin;
diff --git a/src/grpc-clients/src/pool.rs b/src/grpc-clients/src/pool.rs
index 42ed475f7..f9046c371 100644
--- a/src/grpc-clients/src/pool.rs
+++ b/src/grpc-clients/src/pool.rs
@@ -30,42 +30,45 @@ use crate::placement::placement::PlacementServiceManager;
 #[derive(Clone)]
 pub struct ClientPool {
     max_open_connection: u64,
-    // placement center
+    // modules: placement center
     placement_center_inner_pools: DashMap<String, Pool<PlacementServiceManager>>,
     placement_center_journal_service_pools: DashMap<String, Pool<JournalServiceManager>>,
     placement_center_kv_service_pools: DashMap<String, Pool<KvServiceManager>>,
     placement_center_mqtt_service_pools: DashMap<String, Pool<MqttServiceManager>>,
     placement_center_openraft_service_pools: DashMap<String, Pool<OpenRaftServiceManager>>,
+    // modules: placement center service: leader cache
+    placement_center_leader_addr_caches: DashMap<String, String>,
 
-    // mqtt broker
+    // modules: mqtt broker
     mqtt_broker_placement_service_pools: DashMap<String, Pool<MqttBrokerPlacementServiceManager>>,
     mqtt_broker_admin_service_pools: DashMap<String, Pool<MqttBrokerAdminServiceManager>>,
 
-    // journal engine
+    // modules: journal engine
     journal_admin_service_pools: DashMap<String, Pool<JournalAdminServiceManager>>,
     journal_inner_service_pools: DashMap<String, Pool<JournalInnerServiceManager>>,
-
-    // leader cache
-    placement_center_leader_addr_caches: DashMap<String, String>,
 }
 
 impl ClientPool {
     pub fn new(max_open_connection: u64) -> Self {
         Self {
             max_open_connection,
+            // modules: placement_center
             placement_center_inner_pools: DashMap::with_capacity(2),
             placement_center_journal_service_pools: DashMap::with_capacity(2),
             placement_center_kv_service_pools: DashMap::with_capacity(2),
             placement_center_mqtt_service_pools: DashMap::with_capacity(2),
-            mqtt_broker_placement_service_pools: DashMap::with_capacity(2),
-            mqtt_broker_admin_service_pools: DashMap::with_capacity(2),
             placement_center_openraft_service_pools: DashMap::with_capacity(2),
             placement_center_leader_addr_caches: DashMap::with_capacity(2),
+            // modules: mqtt_broker
+            mqtt_broker_placement_service_pools: DashMap::with_capacity(2),
+            mqtt_broker_admin_service_pools: DashMap::with_capacity(2),
+            // modules: journal_engine
             journal_admin_service_pools: DashMap::with_capacity(2),
             journal_inner_service_pools: DashMap::with_capacity(2),
         }
     }
 
+    // ----------modules: placement center -------------
     pub async fn placement_center_inner_services_client(
         &self,
         addr: String,
@@ -202,23 +205,25 @@ impl ClientPool {
         ))
     }
 
-    pub async fn mqtt_broker_mqtt_services_client(
+    pub async fn placement_center_openraft_services_client(
         &self,
         addr: String,
-    ) -> Result<Connection<MqttBrokerPlacementServiceManager>, CommonError> {
-        let module = "BrokerPlacementServices".to_string();
-        let key = format!("{}_{}_{}", "MQTTBroker", module, addr);
-
-        if !self.mqtt_broker_placement_service_pools.contains_key(&key) {
-            let manager = MqttBrokerPlacementServiceManager::new(addr.clone());
+    ) -> Result<Connection<OpenRaftServiceManager>, CommonError> {
+        let module = "OpenRaftServices".to_string();
+        let key = format!("{}_{}_{}", "PlacementCenter", module, addr);
+        if !self
+            .placement_center_openraft_service_pools
+            .contains_key(&key)
+        {
+            let manager = OpenRaftServiceManager::new(addr.clone());
             let pool = Pool::builder()
                 .max_open(self.max_open_connection)
                 .build(manager);
-            self.mqtt_broker_placement_service_pools
+            self.placement_center_openraft_service_pools
                 .insert(key.clone(), pool);
         }
 
-        if let Some(pool) = self.mqtt_broker_placement_service_pools.get(&key) {
+        if let Some(pool) = self.placement_center_openraft_service_pools.get(&key) {
             match pool.get().await {
                 Ok(conn) => {
                     return Ok(conn);
@@ -231,29 +236,31 @@ impl ClientPool {
                 }
             };
         }
+
         Err(CommonError::NoAvailableGrpcConnection(
             module,
             "connection pool is not initialized".to_string(),
         ))
     }
 
-    pub async fn mqtt_broker_admin_services_client(
+    // ----------modules: mqtt broker -------------
+    pub async fn mqtt_broker_mqtt_services_client(
         &self,
         addr: String,
-    ) -> Result<Connection<MqttBrokerAdminServiceManager>, CommonError> {
-        let module = "BrokerAdminServices".to_string();
+    ) -> Result<Connection<MqttBrokerPlacementServiceManager>, CommonError> {
+        let module = "BrokerPlacementServices".to_string();
         let key = format!("{}_{}_{}", "MQTTBroker", module, addr);
 
-        if !self.mqtt_broker_admin_service_pools.contains_key(&key) {
-            let manager = MqttBrokerAdminServiceManager::new(addr.clone());
+        if !self.mqtt_broker_placement_service_pools.contains_key(&key) {
+            let manager = MqttBrokerPlacementServiceManager::new(addr.clone());
             let pool = Pool::builder()
                 .max_open(self.max_open_connection)
                 .build(manager);
-            self.mqtt_broker_admin_service_pools
+            self.mqtt_broker_placement_service_pools
                 .insert(key.clone(), pool);
         }
 
-        if let Some(pool) = self.mqtt_broker_admin_service_pools.get(&key) {
+        if let Some(pool) = self.mqtt_broker_placement_service_pools.get(&key) {
             match pool.get().await {
                 Ok(conn) => {
                     return Ok(conn);
@@ -272,25 +279,23 @@ impl ClientPool {
         ))
     }
 
-    pub async fn placement_center_openraft_services_client(
+    pub async fn mqtt_broker_admin_services_client(
         &self,
         addr: String,
-    ) -> Result<Connection<OpenRaftServiceManager>, CommonError> {
-        let module = "OpenRaftServices".to_string();
-        let key = format!("{}_{}_{}", "PlacementCenter", module, addr);
-        if !self
-            .placement_center_openraft_service_pools
-            .contains_key(&key)
-        {
-            let manager = OpenRaftServiceManager::new(addr.clone());
+    ) -> Result<Connection<MqttBrokerAdminServiceManager>, CommonError> {
+        let module = "BrokerAdminServices".to_string();
+        let key = format!("{}_{}_{}", "MQTTBroker", module, addr);
+
+        if !self.mqtt_broker_admin_service_pools.contains_key(&key) {
+            let manager = MqttBrokerAdminServiceManager::new(addr.clone());
             let pool = Pool::builder()
                 .max_open(self.max_open_connection)
                 .build(manager);
-            self.placement_center_openraft_service_pools
+            self.mqtt_broker_admin_service_pools
                 .insert(key.clone(), pool);
         }
 
-        if let Some(pool) = self.placement_center_openraft_service_pools.get(&key) {
+        if let Some(pool) = self.mqtt_broker_admin_service_pools.get(&key) {
             match pool.get().await {
                 Ok(conn) => {
                     return Ok(conn);
@@ -303,13 +308,13 @@ impl ClientPool {
                 }
             };
         }
-
         Err(CommonError::NoAvailableGrpcConnection(
             module,
             "connection pool is not initialized".to_string(),
         ))
     }
 
+    // ----------modules: journal engine -------------
     pub async fn journal_inner_services_client(
         &self,
         addr: String,
@@ -378,6 +383,7 @@ impl ClientPool {
         ))
     }
 
+    // other
     pub fn get_leader_addr(&self, addr: &str) -> Option<String> {
         if let Some(leader_addr) = self.placement_center_leader_addr_caches.get(addr) {
             return Some(leader_addr.clone());
diff --git a/src/journal-server/src/handler/data.rs b/src/journal-server/src/handler/data.rs
index 5621d5d5f..e09ae344c 100644
--- a/src/journal-server/src/handler/data.rs
+++ b/src/journal-server/src/handler/data.rs
@@ -66,12 +66,12 @@ impl DataHandler {
 
         let req_body = request.body.unwrap();
         for message in req_body.data.iter() {
-            let segment_iden = SegmentIdentity {
+            let segment_identity = SegmentIdentity {
                 namespace: message.namespace.to_string(),
                 shard_name: message.shard_name.to_string(),
                 segment_seq: message.segment,
             };
-            self.valitator(&segment_iden)?;
+            self.validator(&segment_identity)?;
         }
 
         let results = write_data_req(
@@ -96,12 +96,12 @@ impl DataHandler {
         let req_body = request.body.unwrap();
         let conf = journal_server_conf();
         for row in req_body.messages.clone() {
-            let segment_iden = SegmentIdentity {
+            let segment_identity = SegmentIdentity {
                 namespace: row.namespace.to_string(),
                 shard_name: row.shard_name.to_string(),
                 segment_seq: row.segment,
             };
-            self.valitator(&segment_iden)?;
+            self.validator(&segment_identity)?;
         }
 
         let conf = journal_server_conf();
@@ -162,33 +162,33 @@ impl DataHandler {
         Ok(result)
     }
 
-    fn valitator(&self, segment_iden: &SegmentIdentity) -> Result<(), JournalServerError> {
+    fn validator(&self, segment_identity: &SegmentIdentity) -> Result<(), JournalServerError> {
         if self
             .cache_manager
-            .get_shard(&segment_iden.namespace, &segment_iden.shard_name)
+            .get_shard(&segment_identity.namespace, &segment_identity.shard_name)
             .is_none()
         {
             return Err(JournalServerError::ShardNotExist(
-                segment_iden.shard_name.to_string(),
+                segment_identity.shard_name.to_string(),
             ));
         }
 
-        let segment = if let Some(segment) = self.cache_manager.get_segment(segment_iden) {
+        let segment = if let Some(segment) = self.cache_manager.get_segment(segment_identity) {
             segment
         } else {
-            return Err(JournalServerError::SegmentNotExist(segment_iden.name()));
+            return Err(JournalServerError::SegmentNotExist(segment_identity.name()));
         };
 
         if !segment.allow_read() {
             return Err(JournalServerError::SegmentStatusError(
-                segment_iden.name(),
+                segment_identity.name(),
                 segment.status.to_string(),
             ));
         }
 
         let conf = journal_server_conf();
         if segment.leader != conf.node_id {
-            return Err(JournalServerError::NotLeader(segment_iden.name()));
+            return Err(JournalServerError::NotLeader(segment_identity.name()));
         }
 
         Ok(())
diff --git a/src/mqtt-broker/src/lib.rs b/src/mqtt-broker/src/lib.rs
index 203b8f6e0..c716f8ebc 100644
--- a/src/mqtt-broker/src/lib.rs
+++ b/src/mqtt-broker/src/lib.rs
@@ -189,6 +189,7 @@ where
             conf.grpc_port,
             self.cache_manager.clone(),
             self.subscribe_manager.clone(),
+            self.connection_manager.clone(),
             self.client_pool.clone(),
             self.message_storage_adapter.clone(),
         );
diff --git a/src/mqtt-broker/src/server/connection.rs b/src/mqtt-broker/src/server/connection.rs
index 52ec517ac..a9bad7814 100644
--- a/src/mqtt-broker/src/server/connection.rs
+++ b/src/mqtt-broker/src/server/connection.rs
@@ -18,10 +18,11 @@ use std::sync::atomic::AtomicU64;
 
 use log::error;
 use protocol::mqtt::common::MqttProtocol;
+use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 static CONNECTION_ID_BUILD: AtomicU64 = AtomicU64::new(1);
 
-#[derive(Clone, PartialEq, PartialOrd)]
+#[derive(Clone, Serialize, Deserialize, PartialEq, PartialOrd)]
 pub enum NetworkConnectionType {
     Tcp,
     Tls,
@@ -44,12 +45,13 @@ impl fmt::Display for NetworkConnectionType {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, Serialize, Deserialize)]
 pub struct NetworkConnection {
     pub connection_type: NetworkConnectionType,
     pub connection_id: u64,
     pub protocol: Option<MqttProtocol>,
     pub addr: SocketAddr,
+    #[serde(skip_serializing, skip_deserializing)]
     pub connection_stop_sx: Option<mpsc::Sender<bool>>,
 }
 
diff --git a/src/mqtt-broker/src/server/connection_manager.rs b/src/mqtt-broker/src/server/connection_manager.rs
index 456a92a68..78bdaed83 100644
--- a/src/mqtt-broker/src/server/connection_manager.rs
+++ b/src/mqtt-broker/src/server/connection_manager.rs
@@ -65,6 +65,9 @@ impl ConnectionManager {
         connection_id
     }
 
+    pub fn list_connect(&self) -> DashMap<u64, NetworkConnection> {
+        self.connections.clone()
+    }
     pub fn add_tcp_write(
         &self,
         connection_id: u64,
@@ -311,33 +314,33 @@ impl ConnectionManager {
     }
 
     pub fn get_connect(&self, connect_id: u64) -> Option<NetworkConnection> {
-        if let Some(connec) = self.connections.get(&connect_id) {
-            return Some(connec.clone());
+        if let Some(connect) = self.connections.get(&connect_id) {
+            return Some(connect.clone());
         }
         None
     }
 
     pub fn get_connect_protocol(&self, connect_id: u64) -> Option<MqttProtocol> {
-        if let Some(connec) = self.connections.get(&connect_id) {
-            return connec.protocol.clone();
+        if let Some(connect) = self.connections.get(&connect_id) {
+            return connect.protocol.clone();
         }
         None
     }
 
     pub fn set_connect_protocol(&self, connect_id: u64, protocol: u8) {
-        if let Some(mut connec) = self.connections.get_mut(&connect_id) {
+        if let Some(mut connect) = self.connections.get_mut(&connect_id) {
             match protocol {
-                3 => connec.set_protocol(MqttProtocol::Mqtt3),
-                4 => connec.set_protocol(MqttProtocol::Mqtt4),
-                5 => connec.set_protocol(MqttProtocol::Mqtt5),
+                3 => connect.set_protocol(MqttProtocol::Mqtt3),
+                4 => connect.set_protocol(MqttProtocol::Mqtt4),
+                5 => connect.set_protocol(MqttProtocol::Mqtt5),
                 _ => {}
             };
         }
     }
 
     pub fn is_websocket(&self, connect_id: u64) -> bool {
-        if let Some(connec) = self.connections.get(&connect_id) {
-            return connec.connection_type == NetworkConnectionType::WebSocket;
+        if let Some(connect) = self.connections.get(&connect_id) {
+            return connect.connection_type == NetworkConnectionType::WebSocket;
         }
         false
     }
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index 1b36622a6..76eb8a304 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -15,35 +15,45 @@
 use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
+use common_base::tools::serialize_value;
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::user::MqttUser;
 use protocol::broker_mqtt::broker_mqtt_admin::mqtt_broker_admin_service_server::MqttBrokerAdminService;
 use protocol::broker_mqtt::broker_mqtt_admin::{
     ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
-    DeleteUserRequest, ListUserReply, ListUserRequest,
+    DeleteUserRequest, ListConnectionRaw, ListConnectionReply, ListConnectionRequest,
+    ListUserReply, ListUserRequest,
 };
 use tonic::{Request, Response, Status};
 
 use crate::handler::cache::CacheManager;
 use crate::security::AuthDriver;
+use crate::server::connection_manager::ConnectionManager;
 use crate::storage::cluster::ClusterStorage;
 
 pub struct GrpcAdminServices {
     client_pool: Arc<ClientPool>,
     cache_manager: Arc<CacheManager>,
+    connection_manager: Arc<ConnectionManager>,
 }
 
 impl GrpcAdminServices {
-    pub fn new(client_pool: Arc<ClientPool>, cache_manager: Arc<CacheManager>) -> Self {
+    pub fn new(
+        client_pool: Arc<ClientPool>,
+        cache_manager: Arc<CacheManager>,
+        connection_manager: Arc<ConnectionManager>,
+    ) -> Self {
         GrpcAdminServices {
             client_pool,
             cache_manager,
+            connection_manager,
         }
     }
 }
 
 #[tonic::async_trait]
 impl MqttBrokerAdminService for GrpcAdminServices {
+    // --- cluster ---
     async fn cluster_status(
         &self,
         _: Request<ClusterStatusRequest>,
@@ -67,27 +77,7 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         return Ok(Response::new(reply));
     }
 
-    async fn mqtt_broker_list_user(
-        &self,
-        _: Request<ListUserRequest>,
-    ) -> Result<Response<ListUserReply>, Status> {
-        let mut reply = ListUserReply::default();
-
-        let mut user_list = Vec::new();
-        let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
-        match auth_driver.read_all_user().await {
-            Ok(date) => {
-                date.iter()
-                    .for_each(|user| user_list.push(user.value().encode()));
-            }
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
-        }
-        reply.users = user_list;
-        return Ok(Response::new(reply));
-    }
-
+    // --- user ---
     async fn mqtt_broker_create_user(
         &self,
         request: Request<CreateUserRequest>,
@@ -124,4 +114,52 @@ impl MqttBrokerAdminService for GrpcAdminServices {
             }
         }
     }
+
+    async fn mqtt_broker_list_user(
+        &self,
+        _: Request<ListUserRequest>,
+    ) -> Result<Response<ListUserReply>, Status> {
+        let mut reply = ListUserReply::default();
+
+        let mut user_list = Vec::new();
+        let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
+        match auth_driver.read_all_user().await {
+            Ok(date) => {
+                date.iter()
+                    .for_each(|user| user_list.push(user.value().encode()));
+            }
+            Err(e) => {
+                return Err(Status::cancelled(e.to_string()));
+            }
+        }
+        reply.users = user_list;
+        return Ok(Response::new(reply));
+    }
+
+    // --- connection ---
+    async fn mqtt_broker_list_connection(
+        &self,
+        _: Request<ListConnectionRequest>,
+    ) -> Result<Response<ListConnectionReply>, Status> {
+        let mut reply = ListConnectionReply::default();
+        let mut list_connection_raw: Vec<ListConnectionRaw> = Vec::new();
+        for (key, value) in self.connection_manager.list_connect() {
+            if let Some(mqtt_value) = self.cache_manager.connection_info.clone().get(&key) {
+                let mqtt_info = serialize_value(mqtt_value.value())?;
+                let raw = ListConnectionRaw {
+                    connection_id: value.connection_id,
+                    connection_type: value.connection_type.to_string(),
+                    protocol: match value.protocol {
+                        Some(protocol) => protocol.into(),
+                        None => "None".to_string(),
+                    },
+                    source_addr: value.addr.to_string(),
+                    info: mqtt_info,
+                };
+                list_connection_raw.push(raw);
+            }
+        }
+        reply.list_connection_raw = list_connection_raw;
+        Ok(Response::new(reply))
+    }
 }
diff --git a/src/mqtt-broker/src/server/grpc/server.rs b/src/mqtt-broker/src/server/grpc/server.rs
index 50d61c007..3cececf6e 100644
--- a/src/mqtt-broker/src/server/grpc/server.rs
+++ b/src/mqtt-broker/src/server/grpc/server.rs
@@ -24,12 +24,14 @@ use tonic::transport::Server;
 
 use super::inner::GrpcInnerServices;
 use crate::handler::cache::CacheManager;
+use crate::server::connection_manager::ConnectionManager;
 use crate::server::grpc::admin::services::GrpcAdminServices;
 use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub struct GrpcServer<S> {
     port: u32,
     metadata_cache: Arc<CacheManager>,
+    connection_manager: Arc<ConnectionManager>,
     subscribe_manager: Arc<SubscribeManager>,
     client_pool: Arc<ClientPool>,
     message_storage_adapter: Arc<S>,
@@ -43,12 +45,14 @@ where
         port: u32,
         metadata_cache: Arc<CacheManager>,
         subscribe_manager: Arc<SubscribeManager>,
+        connection_manager: Arc<ConnectionManager>,
         client_pool: Arc<ClientPool>,
         message_storage_adapter: Arc<S>,
     ) -> Self {
         Self {
             port,
             metadata_cache,
+            connection_manager,
             subscribe_manager,
             client_pool,
             message_storage_adapter,
@@ -63,8 +67,11 @@ where
             self.client_pool.clone(),
             self.message_storage_adapter.clone(),
         );
-        let admin_handler =
-            GrpcAdminServices::new(self.client_pool.clone(), self.metadata_cache.clone());
+        let admin_handler = GrpcAdminServices::new(
+            self.client_pool.clone(),
+            self.metadata_cache.clone(),
+            self.connection_manager.clone(),
+        );
         Server::builder()
             .add_service(MqttBrokerInnerServiceServer::new(inner_handler))
             .add_service(MqttBrokerAdminServiceServer::new(admin_handler))
diff --git a/src/protocol/src/broker_mqtt/proto/admin.proto b/src/protocol/src/broker_mqtt/proto/admin.proto
index 5b815aaf4..3cfaedc8c 100644
--- a/src/protocol/src/broker_mqtt/proto/admin.proto
+++ b/src/protocol/src/broker_mqtt/proto/admin.proto
@@ -1,15 +1,21 @@
 syntax = "proto3";
 package broker.mqtt.admin;
 service MQTTBrokerAdminService {
+    // cluster
     rpc cluster_status(ClusterStatusRequest) returns(ClusterStatusReply){}
 
+    // user
     rpc mqtt_broker_create_user(CreateUserRequest) returns(CreateUserReply){}
 
     rpc mqtt_broker_delete_user(DeleteUserRequest) returns(DeleteUserReply){}
 
     rpc mqtt_broker_list_user(ListUserRequest) returns(ListUserReply){}
+
+    // connection
+    rpc mqtt_broker_list_connection(ListConnectionRequest) returns(ListConnectionReply){}
 }
 
+// --------- cluster --------
 message ClusterStatusRequest{
 
 }
@@ -18,8 +24,9 @@ message ClusterStatusReply{
     repeated string nodes = 2;
 }
 
+// --------- user --------
 message ListUserRequest{
-    
+
 }
 
 message ListUserReply{
@@ -45,3 +52,20 @@ message DeleteUserRequest{
 message DeleteUserReply{
 
 }
+
+// --------- connection --------
+message ListConnectionRequest{
+
+}
+
+message ListConnectionReply{
+    repeated ListConnectionRaw list_connection_raw = 1;
+}
+
+message ListConnectionRaw {
+    uint64 connection_id = 1;
+    string connection_type = 2;
+    string protocol = 3;
+    string source_addr = 4;
+    string info = 5;
+}


Test patch:
diff --git a/src/grpc-clients/tests/mqtt/mod.rs b/src/grpc-clients/tests/mqtt/mod.rs
index 247b1f680..4b54b3e02 100644
--- a/src/grpc-clients/tests/mqtt/mod.rs
+++ b/src/grpc-clients/tests/mqtt/mod.rs
@@ -13,3 +13,4 @@
 // limitations under the License.
 
 mod mqtt_admin_test;
+mod mqtt_connect_test;
diff --git a/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs b/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs
index 1bf36c761..6ae92d116 100644
--- a/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs
+++ b/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs
@@ -28,7 +28,6 @@ mod tests {
     use crate::common::get_mqtt_broker_addr;
 
     #[tokio::test]
-
     async fn cluster_status_test() {
         let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
         let addrs = vec![get_mqtt_broker_addr()];
@@ -45,7 +44,6 @@ mod tests {
     }
 
     #[tokio::test]
-
     async fn user_test() {
         let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
         let addrs = vec![get_mqtt_broker_addr()];
diff --git a/src/grpc-clients/tests/mqtt/mqtt_connect_test.rs b/src/grpc-clients/tests/mqtt/mqtt_connect_test.rs
new file mode 100644
index 000000000..cb98b9d63
--- /dev/null
+++ b/src/grpc-clients/tests/mqtt/mqtt_connect_test.rs
@@ -0,0 +1,41 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use grpc_clients::mqtt::admin::call::mqtt_broker_list_connection;
+    use grpc_clients::pool::ClientPool;
+    use protocol::broker_mqtt::broker_mqtt_admin::ListConnectionRequest;
+
+    use crate::common::get_mqtt_broker_addr;
+
+    #[tokio::test]
+    async fn test_list_connection() {
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let addrs = vec![get_mqtt_broker_addr()];
+
+        match mqtt_broker_list_connection(client_pool, addrs, ListConnectionRequest {}).await {
+            Ok(data) => {
+                println!("{:?}", data);
+            }
+
+            Err(e) => {
+                eprintln!("Failed to list connections: {:?}", e);
+                std::process::exit(1);
+            }
+        };
+    }
+}