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/src/grpc-clients/src/mqtt/admin/call.rs b/src/grpc-clients/src/mqtt/admin/call.rs
index 811616d84..ea025496e 100644
--- a/src/grpc-clients/src/mqtt/admin/call.rs
+++ b/src/grpc-clients/src/mqtt/admin/call.rs
@@ -16,7 +16,10 @@ use std::sync::Arc;
 
 use common_base::error::common::CommonError;
 use prost::Message as _;
-use protocol::broker_mqtt::broker_mqtt_admin::{ClusterStatusReply, ClusterStatusRequest};
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
+    DeleteUserRequest, ListUserReply, ListUserRequest,
+};
 
 use crate::mqtt::{retry_call, MQTTBrokerPlacementInterface, MQTTBrokerService};
 use crate::poll::ClientPool;
@@ -43,3 +46,72 @@ pub async fn cluster_status(
         Err(e) => Err(e),
     }
 }
+
+pub async fn mqtt_broker_list_user(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: ListUserRequest,
+) -> Result<ListUserReply, CommonError> {
+    let request_date = ListUserRequest::encode_to_vec(&request);
+    match retry_call(
+        MQTTBrokerService::Admin,
+        MQTTBrokerPlacementInterface::ListUser,
+        client_poll,
+        addrs,
+        request_date,
+    )
+    .await
+    {
+        Ok(data) => match ListUserReply::decode(data.as_ref()) {
+            Ok(da) => Ok(da),
+            Err(e) => Err(CommonError::CommmonError(e.to_string())),
+        },
+        Err(e) => Err(e),
+    }
+}
+
+pub async fn mqtt_broker_create_user(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: CreateUserRequest,
+) -> Result<CreateUserReply, CommonError> {
+    let request_date = CreateUserRequest::encode_to_vec(&request);
+    match retry_call(
+        MQTTBrokerService::Admin,
+        MQTTBrokerPlacementInterface::CreateUser,
+        client_poll,
+        addrs,
+        request_date,
+    )
+    .await
+    {
+        Ok(data) => match CreateUserReply::decode(data.as_ref()) {
+            Ok(da) => Ok(da),
+            Err(e) => Err(CommonError::CommmonError(e.to_string())),
+        },
+        Err(e) => Err(e),
+    }
+}
+
+pub async fn mqtt_broker_delete_user(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: DeleteUserRequest,
+) -> Result<DeleteUserReply, CommonError> {
+    let request_date = DeleteUserRequest::encode_to_vec(&request);
+    match retry_call(
+        MQTTBrokerService::Admin,
+        MQTTBrokerPlacementInterface::DeleteUser,
+        client_poll,
+        addrs,
+        request_date,
+    )
+    .await
+    {
+        Ok(data) => match DeleteUserReply::decode(data.as_ref()) {
+            Ok(da) => Ok(da),
+            Err(e) => Err(CommonError::CommmonError(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 c4da1077b..23e0ce6d0 100644
--- a/src/grpc-clients/src/mqtt/admin/inner.rs
+++ b/src/grpc-clients/src/mqtt/admin/inner.rs
@@ -15,7 +15,10 @@
 use common_base::error::common::CommonError;
 use mobc::Connection;
 use prost::Message;
-use protocol::broker_mqtt::broker_mqtt_admin::{ClusterStatusReply, ClusterStatusRequest};
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
+    DeleteUserRequest, ListUserReply, ListUserRequest,
+};
 
 use super::MqttBrokerAdminServiceManager;
 
@@ -31,3 +34,42 @@ pub(crate) async fn inner_cluster_status(
         Err(e) => Err(CommonError::CommmonError(e.to_string())),
     }
 }
+
+pub(crate) async fn inner_list_user(
+    mut client: Connection<MqttBrokerAdminServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, CommonError> {
+    match ListUserRequest::decode(request.as_ref()) {
+        Ok(request) => match client.mqtt_broker_list_user(request).await {
+            Ok(result) => Ok(ListUserReply::encode_to_vec(&result.into_inner())),
+            Err(e) => Err(CommonError::GrpcServerStatus(e)),
+        },
+        Err(e) => Err(CommonError::CommmonError(e.to_string())),
+    }
+}
+
+pub(crate) async fn inner_create_user(
+    mut client: Connection<MqttBrokerAdminServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, CommonError> {
+    match CreateUserRequest::decode(request.as_ref()) {
+        Ok(request) => match client.mqtt_broker_create_user(request).await {
+            Ok(result) => Ok(CreateUserReply::encode_to_vec(&result.into_inner())),
+            Err(e) => Err(CommonError::GrpcServerStatus(e)),
+        },
+        Err(e) => Err(CommonError::CommmonError(e.to_string())),
+    }
+}
+
+pub(crate) async fn inner_delete_user(
+    mut client: Connection<MqttBrokerAdminServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, CommonError> {
+    match DeleteUserRequest::decode(request.as_ref()) {
+        Ok(request) => match client.mqtt_broker_delete_user(request).await {
+            Ok(result) => Ok(DeleteUserReply::encode_to_vec(&result.into_inner())),
+            Err(e) => Err(CommonError::GrpcServerStatus(e)),
+        },
+        Err(e) => Err(CommonError::CommmonError(e.to_string())),
+    }
+}
diff --git a/src/grpc-clients/src/mqtt/admin/mod.rs b/src/grpc-clients/src/mqtt/admin/mod.rs
index 84e6a0f97..93d4c7cf2 100644
--- a/src/grpc-clients/src/mqtt/admin/mod.rs
+++ b/src/grpc-clients/src/mqtt/admin/mod.rs
@@ -15,7 +15,7 @@
 use std::sync::Arc;
 
 use common_base::error::common::CommonError;
-use inner::inner_cluster_status;
+use inner::{inner_cluster_status, inner_create_user, inner_delete_user, inner_list_user};
 use mobc::{Connection, Manager};
 use protocol::broker_mqtt::broker_mqtt_admin::mqtt_broker_admin_service_client::MqttBrokerAdminServiceClient;
 use tonic::transport::Channel;
@@ -48,6 +48,13 @@ pub(crate) async fn admin_interface_call(
                 MQTTBrokerPlacementInterface::ClusterStatus => {
                     inner_cluster_status(client, request).await
                 }
+                MQTTBrokerPlacementInterface::ListUser => inner_list_user(client, request).await,
+                MQTTBrokerPlacementInterface::CreateUser => {
+                    inner_create_user(client, request).await
+                }
+                MQTTBrokerPlacementInterface::DeleteUser => {
+                    inner_delete_user(client, request).await
+                }
                 _ => {
                     return Err(CommonError::CommmonError(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 d33684b3f..f289a556c 100644
--- a/src/grpc-clients/src/mqtt/mod.rs
+++ b/src/grpc-clients/src/mqtt/mod.rs
@@ -39,6 +39,9 @@ pub enum MQTTBrokerPlacementInterface {
 
     // admin
     ClusterStatus,
+    ListUser,
+    CreateUser,
+    DeleteUser,
 }
 
 pub mod admin;
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index c9250746f..efbf809cd 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -16,11 +16,16 @@ use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
 use grpc_clients::poll::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};
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    ClusterStatusReply, ClusterStatusRequest, CreateUserReply, CreateUserRequest, DeleteUserReply,
+    DeleteUserRequest, ListUserReply, ListUserRequest,
+};
 use tonic::{Request, Response, Status};
 
 use crate::storage::cluster::ClusterStorage;
+use crate::storage::user::UserStorage;
 
 pub struct GrpcAdminServices {
     client_poll: Arc<ClientPool>,
@@ -56,4 +61,63 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         reply.nodes = broker_node_list;
         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 user_storage = UserStorage::new(self.client_poll.clone());
+        match user_storage.user_list().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));
+    }
+
+    async fn mqtt_broker_create_user(
+        &self,
+        request: Request<CreateUserRequest>,
+    ) -> Result<Response<CreateUserReply>, Status> {
+        let req = request.into_inner();
+
+        let mqtt_user = MqttUser {
+            username: req.username,
+            password: req.password,
+            is_superuser: req.is_superuser,
+        };
+
+        let user_storage = UserStorage::new(self.client_poll.clone());
+
+        match user_storage.save_user(mqtt_user).await {
+            Ok(_) => return Ok(Response::new(CreateUserReply::default())),
+            Err(e) => {
+                return Err(Status::cancelled(e.to_string()));
+            }
+        }
+    }
+
+    async fn mqtt_broker_delete_user(
+        &self,
+        request: Request<DeleteUserRequest>,
+    ) -> Result<Response<DeleteUserReply>, Status> {
+        let req = request.into_inner();
+
+        let user_storage = UserStorage::new(self.client_poll.clone());
+
+        match user_storage.delete_user(req.username).await {
+            Ok(_) => return Ok(Response::new(DeleteUserReply::default())),
+            Err(e) => {
+                return Err(Status::cancelled(e.to_string()));
+            }
+        }
+    }
 }
diff --git a/src/protocol/src/broker_mqtt/proto/admin.proto b/src/protocol/src/broker_mqtt/proto/admin.proto
index 43534cda8..df831dc2b 100644
--- a/src/protocol/src/broker_mqtt/proto/admin.proto
+++ b/src/protocol/src/broker_mqtt/proto/admin.proto
@@ -2,6 +2,13 @@ syntax = "proto3";
 package broker.mqtt.admin;
 service MQTTBrokerAdminService {
     rpc cluster_status(ClusterStatusRequest) returns(ClusterStatusReply){}
+
+    rpc mqtt_broker_create_user(CreateUserRequest) returns(CreateUserReply){}
+
+    rpc mqtt_broker_delete_user(DeleteUserRequest) returns(DeleteUserReply){}
+
+    rpc mqtt_broker_list_user(ListUserRequest) returns(ListUserReply){}
+    
 }
 
 message ClusterStatusRequest{
@@ -11,3 +18,31 @@ message ClusterStatusReply{
     string cluster_name = 1;
     repeated string nodes = 2;
 }
+
+message ListUserRequest{
+    
+}
+
+message ListUserReply{
+    repeated bytes users = 1;
+}
+
+message CreateUserRequest{
+    string username = 1;
+
+    string password = 2;
+
+    bool is_superuser = 3;
+}
+
+message CreateUserReply{
+
+}
+
+message DeleteUserRequest{
+    string username = 1;
+}
+
+message DeleteUserReply{
+
+}


Test patch:
diff --git a/src/grpc-clients/tests/common.rs b/src/grpc-clients/tests/common.rs
index c47ee088b..2779b002f 100644
--- a/src/grpc-clients/tests/common.rs
+++ b/src/grpc-clients/tests/common.rs
@@ -15,3 +15,7 @@
 pub fn get_placement_addr() -> String {
     "127.0.0.1:1228".to_string()
 }
+
+pub fn get_mqtt_broker_addr() -> String {
+    "127.0.0.1:9981".to_string()
+}
diff --git a/src/grpc-clients/tests/mqtt/mod.rs b/src/grpc-clients/tests/mqtt/mod.rs
index 19b7ce8a8..9d03da4e6 100644
--- a/src/grpc-clients/tests/mqtt/mod.rs
+++ b/src/grpc-clients/tests/mqtt/mod.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 mod mqtt_acl_test;
+mod mqtt_admin_test;
 mod mqtt_blacklist_test;
 mod mqtt_last_will_test;
 mod mqtt_session_test;
diff --git a/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs b/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs
new file mode 100644
index 000000000..a98d1aa64
--- /dev/null
+++ b/src/grpc-clients/tests/mqtt/mqtt_admin_test.rs
@@ -0,0 +1,111 @@
+// 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::{
+        cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user, mqtt_broker_list_user,
+    };
+    use grpc_clients::poll::ClientPool;
+    use metadata_struct::mqtt::user::MqttUser;
+    use protocol::broker_mqtt::broker_mqtt_admin::{
+        ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, ListUserRequest,
+    };
+
+    use crate::common::get_mqtt_broker_addr;
+
+    #[tokio::test]
+    async fn cluster_status_test() {
+        let client_poll: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let addrs = vec![get_mqtt_broker_addr()];
+
+        let request = ClusterStatusRequest {};
+        match cluster_status(client_poll.clone(), addrs.clone(), request).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn user_test() {
+        let client_poll: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let addrs = vec![get_mqtt_broker_addr()];
+        let user_name: String = "user1".to_string();
+        let password: String = "123456".to_string();
+
+        let user = CreateUserRequest {
+            username: user_name.clone(),
+            password: password.clone(),
+            is_superuser: false,
+        };
+
+        match mqtt_broker_create_user(client_poll.clone(), addrs.clone(), user.clone()).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+
+        match mqtt_broker_list_user(client_poll.clone(), addrs.clone(), ListUserRequest {}).await {
+            Ok(data) => {
+                let mut flag = false;
+                for raw in data.users {
+                    let mqtt_user = serde_json::from_slice::<MqttUser>(raw.as_slice()).unwrap();
+                    if user.username == mqtt_user.username {
+                        flag = true;
+                    }
+                }
+                assert!(flag, "user1 has been created");
+            }
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        };
+
+        match mqtt_broker_delete_user(
+            client_poll.clone(),
+            addrs.clone(),
+            DeleteUserRequest {
+                username: user.username.clone(),
+            },
+        )
+        .await
+        {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+
+        match mqtt_broker_list_user(client_poll.clone(), addrs.clone(), ListUserRequest {}).await {
+            Ok(data) => {
+                let mut flag = true;
+                for raw in data.users {
+                    let mqtt_user = serde_json::from_slice::<MqttUser>(raw.as_slice()).unwrap();
+                    if user.username == mqtt_user.username {
+                        flag = false;
+                    }
+                }
+                assert!(flag, "user1 should be deleted");
+            }
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        };
+    }
+}
diff --git a/src/grpc-clients/tests/openraft_test.rs b/src/grpc-clients/tests/openraft_test.rs
index 20bc7614d..18a06a3bf 100644
--- a/src/grpc-clients/tests/openraft_test.rs
+++ b/src/grpc-clients/tests/openraft_test.rs
@@ -35,7 +35,7 @@ mod tests {
 
         let node_id = 4;
         let node = Some(Node {
-            rpc_addr: "127.0.0.0:7654".to_string(),
+            rpc_addr: "127.0.0.1:2228".to_string(),
             node_id: 4,
         });
         let blocking = true;