Repository: robustmq/robustmq (language: rust)

Issue:
MINOR: MQTT Broker supports CreateUser, DeleteUser, ListUser GRPC Admin interfaces
### What type of enhancement is this?

Refactor

### What does the enhancement do?

MQTT Broker supports GRPC Admin supports CreateUser, DeleteUser, and ListUser interfaces

### Implementation challenges

_No response_


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 5390b3fb5..79cffd085 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -732,7 +732,9 @@ dependencies = [
  "bytes",
  "common-base",
  "grpc-clients",
+ "metadata-struct",
  "protocol",
+ "serde_json",
  "thiserror",
 ]
 
diff --git a/src/cli-command/Cargo.toml b/src/cli-command/Cargo.toml
index bd32c2c17..6b4a747ab 100644
--- a/src/cli-command/Cargo.toml
+++ b/src/cli-command/Cargo.toml
@@ -25,4 +25,6 @@ axum.workspace = true
 thiserror.workspace = true
 grpc-clients.workspace = true
 common-base.workspace = true
-protocol.workspace = true
\ No newline at end of file
+metadata-struct.workspace = true
+protocol.workspace = true
+serde_json.workspace = true
\ No newline at end of file
diff --git a/src/cli-command/src/mqtt.rs b/src/cli-command/src/mqtt.rs
index 5c135bcb4..1c6db19a0 100644
--- a/src/cli-command/src/mqtt.rs
+++ b/src/cli-command/src/mqtt.rs
@@ -14,31 +14,59 @@
 
 use std::sync::Arc;
 
-use grpc_clients::mqtt::admin::call::cluster_status;
+use grpc_clients::mqtt::admin::call::{
+    cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user, mqtt_broker_list_user,
+};
 use grpc_clients::pool::ClientPool;
-use protocol::broker_mqtt::broker_mqtt_admin::ClusterStatusRequest;
+use metadata_struct::mqtt::user::MqttUser;
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, ListUserRequest,
+};
 
 use crate::{error_info, grpc_addr};
 
 #[derive(Clone)]
 pub struct MqttCliCommandParam {
     pub server: String,
-    pub action: String,
+    pub action: MqttActionType,
 }
 
+#[derive(Clone)]
 pub enum MqttActionType {
-    STATUS,
+    Status,
+    CreateUser(CreateUserCliRequest),
+    DeleteUser(DeleteUserCliRequest),
+    ListUser,
+}
+
+#[derive(Debug, Clone)]
+pub struct CreateUserCliRequest {
+    username: String,
+    password: String,
+    is_superuser: bool,
 }
 
-impl From<String> for MqttActionType {
-    fn from(s: String) -> Self {
-        match s.as_str() {
-            "status" => MqttActionType::STATUS,
-            _ => panic!("Invalid action type {}", s),
+impl CreateUserCliRequest {
+    pub fn new(username: String, password: String, is_superuser: bool) -> Self {
+        CreateUserCliRequest {
+            username,
+            password,
+            is_superuser,
         }
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct DeleteUserCliRequest {
+    username: String,
+}
+
+impl DeleteUserCliRequest {
+    pub fn new(username: String) -> Self {
+        DeleteUserCliRequest { username }
+    }
+}
+
 pub struct MqttBrokerCommand {}
 
 impl Default for MqttBrokerCommand {
@@ -53,12 +81,22 @@ impl MqttBrokerCommand {
     }
 
     pub async fn start(&self, params: MqttCliCommandParam) {
-        let action_type = MqttActionType::from(params.action.clone());
         let client_pool = Arc::new(ClientPool::new(100));
-        match action_type {
-            MqttActionType::STATUS => {
+        match params.action {
+            MqttActionType::Status => {
                 self.status(client_pool.clone(), params.clone()).await;
             }
+            MqttActionType::CreateUser(ref request) => {
+                self.create_user(client_pool.clone(), params.clone(), request.clone())
+                    .await;
+            }
+            MqttActionType::DeleteUser(ref request) => {
+                self.delete_user(client_pool.clone(), params.clone(), request.clone())
+                    .await;
+            }
+            MqttActionType::ListUser => {
+                self.list_user(client_pool.clone(), params.clone()).await;
+            }
         }
     }
 
@@ -79,4 +117,65 @@ impl MqttBrokerCommand {
             }
         }
     }
+
+    async fn create_user(
+        &self,
+        client_pool: Arc<ClientPool>,
+        params: MqttCliCommandParam,
+        cli_request: CreateUserCliRequest,
+    ) {
+        let request = CreateUserRequest {
+            username: cli_request.username,
+            password: cli_request.password,
+            is_superuser: cli_request.is_superuser,
+        };
+        match mqtt_broker_create_user(client_pool.clone(), grpc_addr(params.server), request).await
+        {
+            Ok(_) => {
+                println!("Created successfully!",)
+            }
+            Err(e) => {
+                println!("MQTT broker create user normal exception");
+                error_info(e.to_string());
+            }
+        }
+    }
+
+    async fn delete_user(
+        &self,
+        client_pool: Arc<ClientPool>,
+        params: MqttCliCommandParam,
+        cli_request: DeleteUserCliRequest,
+    ) {
+        let request = DeleteUserRequest {
+            username: cli_request.username,
+        };
+        match mqtt_broker_delete_user(client_pool.clone(), grpc_addr(params.server), request).await
+        {
+            Ok(_) => {
+                println!("Deleted successfully!");
+            }
+            Err(e) => {
+                println!("MQTT broker delete user normal exception");
+                error_info(e.to_string());
+            }
+        }
+    }
+
+    async fn list_user(&self, client_pool: Arc<ClientPool>, params: MqttCliCommandParam) {
+        let request = ListUserRequest {};
+        match mqtt_broker_list_user(client_pool.clone(), grpc_addr(params.server), request).await {
+            Ok(data) => {
+                println!("user list:");
+                for user in data.users {
+                    let mqtt_user = serde_json::from_slice::<MqttUser>(user.as_slice()).unwrap();
+                    println!("{},", mqtt_user.username);
+                }
+            }
+            Err(e) => {
+                println!("MQTT broker list user exception");
+                error_info(e.to_string());
+            }
+        }
+    }
 }
diff --git a/src/cmd/src/cli-command/command.rs b/src/cmd/src/cli-command/command.rs
index f6c70f81b..eab1e135e 100644
--- a/src/cmd/src/cli-command/command.rs
+++ b/src/cmd/src/cli-command/command.rs
@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use clap::{Parser, Subcommand};
-use cli_command::mqtt::{MqttBrokerCommand, MqttCliCommandParam};
+use cli_command::mqtt::{
+    CreateUserCliRequest, DeleteUserCliRequest, MqttActionType, MqttBrokerCommand,
+    MqttCliCommandParam,
+};
 use cli_command::placement::{
     AddLearnerCliRequset, ChangeMembershipCliRequest, PlacementActionType, PlacementCenterCommand,
     PlacementCliCommandParam,
@@ -47,15 +50,45 @@ struct MqttArgs {
     #[arg(short, long,default_value_t =String::from("127.0.0.1:9981"))]
     server: String,
 
-    #[arg(short, long,default_value_t =String::from("status"))]
-    action: String,
+    #[clap(subcommand)]
+    action: MQTTAction,
+}
+
+#[derive(Debug, Subcommand)]
+enum MQTTAction {
+    Status,
+    CreateUser(CreateUserArgs),
+    DeleteUser(DeleteUserArgs),
+    ListUser,
+}
+
+#[derive(clap::Args, Debug)]
+#[command(author="RobustMQ", about="action: create user", long_about = None)]
+#[command(next_line_help = true)]
+struct CreateUserArgs {
+    #[arg(short, long, required = true)]
+    username: String,
+
+    #[arg(short, long, required = true)]
+    password: String,
+
+    #[arg(short, long, default_value_t = false)]
+    is_superuser: bool,
+}
+
+#[derive(clap::Args, Debug)]
+#[command(author="RobustMQ", about="action: delete user", long_about = None)]
+#[command(next_line_help = true)]
+struct DeleteUserArgs {
+    #[arg(short, long, required = true)]
+    username: String,
 }
 
 #[derive(clap::Args, Debug)]
 #[command(author="RobustMQ",  about="Command line tool for placement center", long_about = None)]
 #[command(next_line_help = true)]
 struct PlacementArgs {
-    #[arg(short, long, default_value_t =String::from("127.0.0.1:1228"))]
+    #[arg(short, long, default_value_t = String::from("127.0.0.1:1228"))]
     server: String,
 
     #[clap(subcommand)]
@@ -113,7 +146,16 @@ async fn main() {
             let cmd = MqttBrokerCommand::new();
             let params = MqttCliCommandParam {
                 server: args.server,
-                action: args.action,
+                action: match args.action {
+                    MQTTAction::Status => MqttActionType::Status,
+                    MQTTAction::CreateUser(arg) => MqttActionType::CreateUser(
+                        CreateUserCliRequest::new(arg.username, arg.password, arg.is_superuser),
+                    ),
+                    MQTTAction::DeleteUser(arg) => {
+                        MqttActionType::DeleteUser(DeleteUserCliRequest::new(arg.username))
+                    }
+                    MQTTAction::ListUser => MqttActionType::ListUser,
+                },
             };
             cmd.start(params).await;
         }


Test patch: