Repository: robustmq/robustmq (language: rust)

Issue:
COMMAND: MQTT Command supports viewing a list of mqtt topics.
### What problem does the new feature solve?

[MQTT Command supports viewing a list of mqtt topics.](https://docs.emqx.com/zh/enterprise/v5.8/admin/api-docs.html#tag/Topics/paths/~1topics~1%7Btopic%7D/get)

### What does the feature do?

MQTT Command supports viewing a list of mqtt topics.

### Implementation challenges

_No response_


Discussion / hints:
mark


Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 633f0582e..a6f693d7b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -757,6 +757,7 @@ dependencies = [
  "common-base",
  "journal-server",
  "lazy_static",
+ "mockall",
  "mqtt-broker",
  "placement-center",
  "protocol",
@@ -1051,6 +1052,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "downcast"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1"
+
 [[package]]
 name = "dunce"
 version = "1.0.4"
@@ -1151,6 +1158,12 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fragile"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"
+
 [[package]]
 name = "frunk"
 version = "0.4.2"
@@ -2105,6 +2118,32 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "mockall"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "39a6bfcc6c8c7eed5ee98b9c3e33adc726054389233e201c95dab2d41a3839d2"
+dependencies = [
+ "cfg-if",
+ "downcast",
+ "fragile",
+ "mockall_derive",
+ "predicates",
+ "predicates-tree",
+]
+
+[[package]]
+name = "mockall_derive"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25ca3004c2efe9011bd4e461bd8256445052b9615405b4f7ea43fc8ca5c20898"
+dependencies = [
+ "cfg-if",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.48",
+]
+
 [[package]]
 name = "mqtt-bridge-elasticsearch"
 version = "0.1.1"
@@ -2691,6 +2730,32 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
 
+[[package]]
+name = "predicates"
+version = "3.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e9086cc7640c29a356d1a29fd134380bee9d8f79a17410aa76e7ad295f42c97"
+dependencies = [
+ "anstyle",
+ "predicates-core",
+]
+
+[[package]]
+name = "predicates-core"
+version = "1.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae8177bee8e75d6846599c6b9ff679ed51e882816914eec639944d7c9aa11931"
+
+[[package]]
+name = "predicates-tree"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "41b740d195ed3166cd147c8047ec98db0e22ec019eb8eeb76d343b795304fb13"
+dependencies = [
+ "predicates-core",
+ "termtree",
+]
+
 [[package]]
 name = "prettyplease"
 version = "0.1.25"
@@ -3704,6 +3769,12 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "termtree"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"
+
 [[package]]
 name = "third-driver"
 version = "0.1.1"
diff --git a/Cargo.toml b/Cargo.toml
index 8075b754b..3a4c6cb19 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -98,6 +98,9 @@ rand = "0.8.5"
 clap-cargo = "0.14.1"
 clap = { version = "4.4.7", features = ["derive"] }
 
+## unit test framework
+mockall = "0.13.1"
+
 
 ## workspaces members
 mqtt-bridge-kafka = { path = "src/mqtt-bridge/kafka" }
@@ -121,7 +124,6 @@ third-driver = { path = "src/common/third-driver" }
 protocol = { path = "src/protocol" }
 robustmq-test = { path = "tests" }
 
-
 [profile.dev]
 overflow-checks = false
 incremental = true
diff --git a/src/cli-command/src/mqtt.rs b/src/cli-command/src/mqtt.rs
index 88a13c937..80f765747 100644
--- a/src/cli-command/src/mqtt.rs
+++ b/src/cli-command/src/mqtt.rs
@@ -16,13 +16,14 @@ use std::sync::Arc;
 
 use grpc_clients::mqtt::admin::call::{
     cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user,
-    mqtt_broker_enable_slow_subscribe, mqtt_broker_list_connection, mqtt_broker_list_user,
+    mqtt_broker_enable_slow_subscribe, mqtt_broker_list_connection, mqtt_broker_list_topic,
+    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, EnableSlowSubscribeRequest,
-    ListConnectionRequest, ListUserRequest,
+    ListConnectionRequest, ListTopicRequest, ListUserRequest,
 };
 
 use crate::{error_info, grpc_addr};
@@ -47,6 +48,8 @@ pub enum MqttActionType {
 
     // observability: slow-ub
     EnableSlowSubscribe(EnableSlowSubscribeRequest),
+
+    ListTopic(ListTopicRequest),
 }
 
 pub struct MqttBrokerCommand {}
@@ -87,6 +90,10 @@ impl MqttBrokerCommand {
                 self.enable_slow_subscribe(client_pool.clone(), params.clone(), request.clone())
                     .await;
             }
+            MqttActionType::ListTopic(ref request) => {
+                self.list_topic(client_pool.clone(), params.clone(), request.clone())
+                    .await;
+            }
         }
     }
 
@@ -210,4 +217,37 @@ impl MqttBrokerCommand {
             }
         }
     }
+
+    async fn list_topic(
+        &self,
+        client_pool: Arc<ClientPool>,
+        params: MqttCliCommandParam,
+        cli_request: ListTopicRequest,
+    ) {
+        match mqtt_broker_list_topic(client_pool.clone(), &grpc_addr(params.server), cli_request)
+            .await
+        {
+            Ok(data) => {
+                println!("topic list result:");
+                for mqtt_topic in data.topics {
+                    println!(
+                        concat!(
+                            "topic id: {}\n",
+                            "topic name: {}\n",
+                            "cluster name: {}\n",
+                            "is contain retain message: {}\n"
+                        ),
+                        mqtt_topic.topic_id,
+                        mqtt_topic.topic_name,
+                        mqtt_topic.cluster_name,
+                        mqtt_topic.is_contain_retain_message
+                    );
+                }
+            }
+            Err(e) => {
+                println!("MQTT broker list topic exception");
+                error_info(e.to_string());
+            }
+        }
+    }
 }
diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml
index fca4b6e98..03ae8ecc5 100644
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -46,3 +46,6 @@ journal-server.workspace = true
 cli-command.workspace = true
 clap-cargo.workspace = true
 protocol.workspace = true
+
+[dev-dependencies]
+mockall.workspace = true
\ No newline at end of file
diff --git a/src/cmd/src/cli-command/command.rs b/src/cmd/src/cli-command/command.rs
index cf3491837..e411c8390 100644
--- a/src/cmd/src/cli-command/command.rs
+++ b/src/cmd/src/cli-command/command.rs
@@ -15,12 +15,14 @@
 pub(crate) mod mqtt;
 mod utils;
 
-use clap::{arg, Parser, Subcommand};
+use clap::{arg, Parser, Subcommand, ValueEnum};
 use cli_command::mqtt::{MqttActionType, MqttBrokerCommand, MqttCliCommandParam};
 use cli_command::placement::{
     PlacementActionType, PlacementCenterCommand, PlacementCliCommandParam,
 };
-use protocol::broker_mqtt::broker_mqtt_admin::{CreateUserRequest, DeleteUserRequest};
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    CreateUserRequest, DeleteUserRequest, ListTopicRequest,
+};
 use protocol::placement_center::placement_center_openraft::{
     AddLearnerRequest, ChangeMembershipRequest, Node,
 };
@@ -70,11 +72,31 @@ enum MQTTAction {
     // Connections
     ListConnection,
 
+    ListTopic(ListTopicArgs),
+
     // observability: slow-sub feat
     #[clap(name = "slow-sub")]
     SlowSub(SlowSubArgs),
 }
 
+#[derive(ValueEnum, Clone, Debug)]
+enum MatchOption {
+    E,
+    P,
+    S,
+}
+
+#[derive(clap::Args, Debug)]
+#[command(author="RobustMQ", about="action: list topics", long_about = None)]
+#[command(next_line_help = true)]
+struct ListTopicArgs {
+    #[arg(short, long, required = true)]
+    topic_name: String,
+
+    #[arg(short, long, default_value = "e")]
+    match_option: MatchOption,
+}
+
 #[derive(clap::Args, Debug)]
 #[command(author="RobustMQ",  about="Command line tool for placement center", long_about = None)]
 #[command(next_line_help = true)]
@@ -133,55 +155,68 @@ struct JournalArgs {
 async fn main() {
     let args = RobustMQCli::parse();
     match args {
-        RobustMQCli::Mqtt(args) => {
-            let cmd = MqttBrokerCommand::new();
-            let params = MqttCliCommandParam {
-                server: args.server,
-                action: match args.action {
-                    MQTTAction::Status => MqttActionType::Status,
-                    MQTTAction::CreateUser(arg) => MqttActionType::CreateUser(CreateUserRequest {
-                        username: arg.username,
-                        password: arg.password,
-                        is_superuser: arg.is_superuser,
-                    }),
-                    MQTTAction::DeleteUser(arg) => MqttActionType::DeleteUser(DeleteUserRequest {
-                        username: arg.username,
-                    }),
-                    MQTTAction::ListUser => MqttActionType::ListUser,
-                    MQTTAction::ListConnection => MqttActionType::ListConnection,
-                    _ => unreachable!("UnSupport command"),
-                },
-            };
-            cmd.start(params).await;
-        }
-        RobustMQCli::Place(args) => {
-            let cmd = PlacementCenterCommand::new();
-            let params = PlacementCliCommandParam {
-                server: args.server,
-                action: match args.action {
-                    PlacementAction::Status => PlacementActionType::Status,
-                    PlacementAction::AddLearner(arg) => {
-                        PlacementActionType::AddLearner(AddLearnerRequest {
-                            node_id: arg.node_id,
-                            node: Some(Node {
-                                node_id: arg.node_id,
-                                rpc_addr: arg.rpc_addr,
-                            }),
-                            blocking: arg.blocking,
-                        })
-                    }
-                    PlacementAction::ChangeMembership(arg) => {
-                        PlacementActionType::ChangeMembership(ChangeMembershipRequest {
-                            members: arg.members,
-                            retain: arg.retain,
-                        })
-                    }
-                },
-            };
-            cmd.start(params).await;
-        }
-        RobustMQCli::Journal(args) => {
-            println!("{:?}", args);
-        }
+        RobustMQCli::Mqtt(args) => handle_mqtt(args, MqttBrokerCommand::new()).await,
+        RobustMQCli::Place(args) => handle_placement(args, PlacementCenterCommand::new()).await,
+        RobustMQCli::Journal(args) => handle_journal(args).await,
     }
 }
+
+async fn handle_mqtt(args: MqttArgs, cmd: MqttBrokerCommand) {
+    let params = MqttCliCommandParam {
+        server: args.server,
+        action: match args.action {
+            MQTTAction::Status => MqttActionType::Status,
+            MQTTAction::CreateUser(arg) => MqttActionType::CreateUser(CreateUserRequest {
+                username: arg.username,
+                password: arg.password,
+                is_superuser: arg.is_superuser,
+            }),
+            MQTTAction::DeleteUser(arg) => MqttActionType::DeleteUser(DeleteUserRequest {
+                username: arg.username,
+            }),
+            MQTTAction::ListUser => MqttActionType::ListUser,
+            MQTTAction::ListConnection => MqttActionType::ListConnection,
+            MQTTAction::ListTopic(args) => MqttActionType::ListTopic(ListTopicRequest {
+                topic_name: args.topic_name,
+                match_option: match args.match_option {
+                    MatchOption::E => 0,
+                    MatchOption::P => 1,
+                    MatchOption::S => 2,
+                },
+            }),
+            _ => unreachable!("UnSupport command"),
+        },
+    };
+    cmd.start(params).await;
+}
+
+async fn handle_placement(args: PlacementArgs, cmd: PlacementCenterCommand) {
+    let params = PlacementCliCommandParam {
+        server: args.server,
+        action: match args.action {
+            PlacementAction::Status => PlacementActionType::Status,
+            PlacementAction::AddLearner(arg) => {
+                PlacementActionType::AddLearner(AddLearnerRequest {
+                    node_id: arg.node_id,
+                    node: Some(Node {
+                        node_id: arg.node_id,
+                        rpc_addr: arg.rpc_addr,
+                    }),
+                    blocking: arg.blocking,
+                })
+            }
+            PlacementAction::ChangeMembership(arg) => {
+                PlacementActionType::ChangeMembership(ChangeMembershipRequest {
+                    members: arg.members,
+                    retain: arg.retain,
+                })
+            }
+        },
+    };
+    cmd.start(params).await;
+}
+
+// TODO: implement journal engine
+async fn handle_journal(args: JournalArgs) {
+    println!("{:?}", args);
+}
diff --git a/src/grpc-clients/src/mqtt/admin/call.rs b/src/grpc-clients/src/mqtt/admin/call.rs
index 0323e3ae9..1901d7898 100644
--- a/src/grpc-clients/src/mqtt/admin/call.rs
+++ b/src/grpc-clients/src/mqtt/admin/call.rs
@@ -21,7 +21,7 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     DeleteAclReply, DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest,
     DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
     ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionReply,
-    ListConnectionRequest, ListUserReply, ListUserRequest,
+    ListConnectionRequest, ListTopicReply, ListTopicRequest, ListUserReply, ListUserRequest,
 };
 
 use crate::mqtt::{call_once, MqttBrokerPlacementReply, MqttBrokerPlacementRequest};
@@ -177,3 +177,15 @@ pub async fn mqtt_broker_enable_slow_subscribe(
         _ => unreachable!("Reply type mismatch"),
     }
 }
+
+pub async fn mqtt_broker_list_topic(
+    client_pool: Arc<ClientPool>,
+    addrs: &[String],
+    request: ListTopicRequest,
+) -> Result<ListTopicReply, CommonError> {
+    let request = MqttBrokerPlacementRequest::ListTopic(request);
+    match retry_call(&client_pool, addrs, request, call_once).await? {
+        MqttBrokerPlacementReply::ListTopic(reply) => Ok(reply),
+        _ => unreachable!("Reply type mismatch"),
+    }
+}
diff --git a/src/grpc-clients/src/mqtt/mod.rs b/src/grpc-clients/src/mqtt/mod.rs
index fb0773559..b3de443d7 100644
--- a/src/grpc-clients/src/mqtt/mod.rs
+++ b/src/grpc-clients/src/mqtt/mod.rs
@@ -19,7 +19,7 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     DeleteAclReply, DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest,
     DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
     ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionReply,
-    ListConnectionRequest, ListUserReply, ListUserRequest,
+    ListConnectionRequest, ListTopicReply, ListTopicRequest, ListUserReply, ListUserRequest,
 };
 use protocol::broker_mqtt::broker_mqtt_inner::{
     DeleteSessionReply, DeleteSessionRequest, SendLastWillMessageReply, SendLastWillMessageRequest,
@@ -53,6 +53,8 @@ pub enum MqttBrokerPlacementRequest {
 
     // slow subscribe
     EnableSlowSubscribe(EnableSlowSubscribeRequest),
+
+    ListTopic(ListTopicRequest),
 }
 
 /// Enum wrapper for all possible replies from the mqtt broker
@@ -80,6 +82,8 @@ pub enum MqttBrokerPlacementReply {
 
     // slow subscribe
     EnableSlowSubscribe(EnableSlowSubScribeReply),
+
+    ListTopic(ListTopicReply),
 }
 
 pub mod admin;
@@ -179,6 +183,7 @@ async fn call_once(
                 .await?;
             Ok(MqttBrokerPlacementReply::ListConnection(reply.into_inner()))
         }
+
         EnableSlowSubscribe(enable_slow_subscribe_request) => {
             let mut client = client_pool.mqtt_broker_admin_services_client(addr).await?;
             let reply = client
@@ -188,6 +193,12 @@ async fn call_once(
                 reply.into_inner(),
             ))
         }
+
+        ListTopic(list_topic_request) => {
+            let mut client = client_pool.mqtt_broker_admin_services_client(addr).await?;
+            let reply = client.mqtt_broker_list_topic(list_topic_request).await?;
+            Ok(MqttBrokerPlacementReply::ListTopic(reply.into_inner()))
+        }
     }
 }
 
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index 2691b4a09..d7a43b21e 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -27,7 +27,8 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     DeleteAclReply, DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest,
     DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
     ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionRaw,
-    ListConnectionReply, ListConnectionRequest, ListUserReply, ListUserRequest,
+    ListConnectionReply, ListConnectionRequest, ListTopicReply, ListTopicRequest, ListUserReply,
+    ListUserRequest, MqttTopic,
 };
 use tonic::{Request, Response, Status};
 
@@ -312,4 +313,48 @@ impl MqttBrokerAdminService for GrpcAdminServices {
             Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
+
+    async fn mqtt_broker_list_topic(
+        &self,
+        request: Request<ListTopicRequest>,
+    ) -> Result<Response<ListTopicReply>, Status> {
+        let req = request.into_inner();
+        let topic_query_result: Vec<MqttTopic> = match req.match_option {
+            0 => self
+                .cache_manager
+                .get_topic_by_name(&req.topic_name)
+                .into_iter()
+                .take(10)
+                .map(|entry| MqttTopic {
+                    topic_id: entry.topic_id.clone(),
+                    topic_name: entry.topic_name.clone(),
+                    cluster_name: entry.cluster_name.clone(),
+                    is_contain_retain_message: entry.retain_message.is_some(),
+                })
+                .collect(),
+            option => self
+                .cache_manager
+                .topic_info
+                .iter()
+                .filter(|entry| match option {
+                    1 => entry.value().topic_name.starts_with(&req.topic_name),
+                    2 => entry.value().topic_name.contains(&req.topic_name),
+                    _ => false,
+                })
+                .take(10)
+                .map(|entry| MqttTopic {
+                    topic_id: entry.value().topic_id.clone(),
+                    topic_name: entry.value().topic_name.clone(),
+                    cluster_name: entry.value().cluster_name.clone(),
+                    is_contain_retain_message: entry.value().retain_message.is_some(),
+                })
+                .collect(),
+        };
+
+        let reply = ListTopicReply {
+            topics: topic_query_result,
+        };
+
+        return Ok(Response::new(reply));
+    }
 }
diff --git a/src/protocol/src/broker_mqtt/proto/admin.proto b/src/protocol/src/broker_mqtt/proto/admin.proto
index e9ff8b934..9cb365d66 100644
--- a/src/protocol/src/broker_mqtt/proto/admin.proto
+++ b/src/protocol/src/broker_mqtt/proto/admin.proto
@@ -30,27 +30,28 @@ service MQTTBrokerAdminService {
 
     // observability: slow-sub
     rpc mqtt_broker_enable_slow_subscribe(EnableSlowSubscribeRequest) returns(EnableSlowSubScribeReply) {}
+    rpc mqtt_broker_list_topic(ListTopicRequest) returns(ListTopicReply){}
 }
 
 // --------- cluster --------
-message ClusterStatusRequest{
+message ClusterStatusRequest {
 
 }
-message ClusterStatusReply{
+message ClusterStatusReply {
     string cluster_name = 1;
     repeated string nodes = 2;
 }
 
 // --------- user --------
-message ListUserRequest{
+message ListUserRequest {
 
 }
 
-message ListUserReply{
+message ListUserReply {
     repeated bytes users = 1;
 }
 
-message CreateUserRequest{
+message CreateUserRequest {
     string username = 1;
 
     string password = 2;
@@ -58,15 +59,15 @@ message CreateUserRequest{
     bool is_superuser = 3;
 }
 
-message CreateUserReply{
+message CreateUserReply {
 
 }
 
-message DeleteUserRequest{
+message DeleteUserRequest {
     string username = 1;
 }
 
-message DeleteUserReply{
+message DeleteUserReply {
 
 }
 
@@ -129,11 +130,11 @@ message DeleteBlacklistReply{
 }
 
 // --------- connection --------
-message ListConnectionRequest{
+message ListConnectionRequest {
 
 }
 
-message ListConnectionReply{
+message ListConnectionReply {
     repeated ListConnectionRaw list_connection_raw = 1;
 }
 
@@ -145,7 +146,6 @@ message ListConnectionRaw {
     string info = 5;
 }
 
-
 // --------- observability service --------
 // --------- slow subscribe feature -------
 message EnableSlowSubscribeRequest {
@@ -155,3 +155,24 @@ message EnableSlowSubscribeRequest {
 message EnableSlowSubScribeReply {
     bool is_enable = 1;
 }
+
+// --------- mqtt topic --------
+enum MatchOption {
+    E = 0;
+    P = 1;
+    S = 2;
+}
+
+message ListTopicRequest {
+    string topic_name = 1;
+    MatchOption match_option = 2;
+}
+message ListTopicReply {
+    repeated MqttTopic topics = 1;
+}
+message MqttTopic {
+    string topic_id = 1;
+    string cluster_name = 2;
+    string topic_name = 3;
+    bool is_contain_retain_message = 4;
+}


Test patch: