Repository: robustmq/robustmq (language: rust)

Issue:
MINOR: After successful creation of MQTT user by Placement Center, the cache of MQTT Broker is updated
### What type of enhancement is this?

Refactor

### What does the enhancement do?

After successful creation of MQTT user by Placement Center, the cache of MQTT Broker is updated

### Implementation challenges

_No response_


Discussion / hints:

Fix patch:
diff --git a/src/mqtt-broker/src/handler/cache.rs b/src/mqtt-broker/src/handler/cache.rs
index 98804241b..d6d0c631d 100644
--- a/src/mqtt-broker/src/handler/cache.rs
+++ b/src/mqtt-broker/src/handler/cache.rs
@@ -245,7 +245,10 @@ impl CacheManager {
                     let user: MqttUser = serde_json::from_str(&data.value).unwrap();
                     self.add_user(user);
                 }
-                MetadataCacheAction::Del => self.del_user(data.value),
+                MetadataCacheAction::Del => {
+                    let user: MqttUser = serde_json::from_str(&data.value).unwrap();
+                    self.del_user(user.username);
+                }
             },
             MetadataCacheType::Topic => match data.action {
                 MetadataCacheAction::Set => {}
@@ -273,9 +276,8 @@ impl CacheManager {
         self.user_info.insert(user.username.clone(), user);
     }
 
-    pub fn del_user(&self, value: String) {
-        let data: MqttUser = serde_json::from_str(&value).unwrap();
-        self.user_info.remove(&data.username);
+    pub fn del_user(&self, username: String) {
+        self.user_info.remove(&username);
     }
 
     pub fn add_session(&self, client_id: String, session: MqttSession) {
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index 0a7a35491..8dce1f664 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -24,16 +24,21 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
 };
 use tonic::{Request, Response, Status};
 
+use crate::handler::cache::CacheManager;
 use crate::storage::cluster::ClusterStorage;
 use crate::storage::user::UserStorage;
 
 pub struct GrpcAdminServices {
     client_pool: Arc<ClientPool>,
+    cache_manager: Arc<CacheManager>,
 }
 
 impl GrpcAdminServices {
-    pub fn new(client_pool: Arc<ClientPool>) -> Self {
-        GrpcAdminServices { client_pool }
+    pub fn new(client_pool: Arc<ClientPool>, cache_manager: Arc<CacheManager>) -> Self {
+        GrpcAdminServices {
+            client_pool,
+            cache_manager,
+        }
     }
 }
 
@@ -88,7 +93,7 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         request: Request<CreateUserRequest>,
     ) -> Result<Response<CreateUserReply>, Status> {
         let req = request.into_inner();
-
+        let username = req.username.clone();
         let mqtt_user = MqttUser {
             username: req.username,
             password: req.password,
@@ -97,12 +102,28 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let user_storage = UserStorage::new(self.client_pool.clone());
 
-        match user_storage.save_user(mqtt_user).await {
-            Ok(_) => return Ok(Response::new(CreateUserReply::default())),
+        match user_storage.user_list().await {
+            Ok(date) => {
+                let is_existed = date.iter().any(|user| *user.key() == username);
+                if is_existed {
+                    return Err(Status::cancelled("user has beed existed"));
+                }
+            }
+            Err(e) => {
+                return Err(Status::cancelled(e.to_string()));
+            }
+        }
+
+        match user_storage.save_user(mqtt_user.clone()).await {
+            Ok(_) => {}
             Err(e) => {
                 return Err(Status::cancelled(e.to_string()));
             }
         }
+
+        self.cache_manager.add_user(mqtt_user.clone());
+
+        return Ok(Response::new(CreateUserReply::default()));
     }
 
     async fn mqtt_broker_delete_user(
@@ -110,10 +131,25 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         request: Request<DeleteUserRequest>,
     ) -> Result<Response<DeleteUserReply>, Status> {
         let req = request.into_inner();
+        let username = req.username;
 
         let user_storage = UserStorage::new(self.client_pool.clone());
 
-        match user_storage.delete_user(req.username).await {
+        match user_storage.user_list().await {
+            Ok(date) => {
+                let is_existed = date.iter().any(|user| *user.key() == username);
+                if !is_existed {
+                    return Err(Status::cancelled("user does not existed"));
+                }
+            }
+            Err(e) => {
+                return Err(Status::cancelled(e.to_string()));
+            }
+        }
+
+        self.cache_manager.del_user(username.clone());
+
+        match user_storage.delete_user(username.clone()).await {
             Ok(_) => return Ok(Response::new(DeleteUserReply::default())),
             Err(e) => {
                 return Err(Status::cancelled(e.to_string()));
diff --git a/src/mqtt-broker/src/server/grpc/server.rs b/src/mqtt-broker/src/server/grpc/server.rs
index a3b1d5302..dd79411a9 100644
--- a/src/mqtt-broker/src/server/grpc/server.rs
+++ b/src/mqtt-broker/src/server/grpc/server.rs
@@ -63,7 +63,8 @@ where
             self.client_pool.clone(),
             self.message_storage_adapter.clone(),
         );
-        let admin_handler = GrpcAdminServices::new(self.client_pool.clone());
+        let admin_handler =
+            GrpcAdminServices::new(self.client_pool.clone(), self.metadata_cache.clone());
         Server::builder()
             .add_service(MqttBrokerPlacementServiceServer::new(placement_handler))
             .add_service(MqttBrokerAdminServiceServer::new(admin_handler))


Test patch:
diff --git a/src/mqtt-broker/tests/common.rs b/src/mqtt-broker/tests/common.rs
index 486d708e0..6725d94f1 100644
--- a/src/mqtt-broker/tests/common.rs
+++ b/src/mqtt-broker/tests/common.rs
@@ -39,6 +39,10 @@ pub fn broker_ws_addr() -> String {
 pub fn broker_wss_addr() -> String {
     "wss://127.0.0.1:8094".to_string()
 }
+#[allow(dead_code)]
+pub fn broker_grpc_addr() -> String {
+    "127.0.0.1:9981".to_string()
+}
 
 #[allow(dead_code)]
 pub fn username() -> String {
@@ -113,6 +117,41 @@ pub fn build_v5_conn_pros(props: Properties, err_pwd: bool, ws: bool, ssl: bool)
         .finalize()
 }
 
+#[allow(dead_code)]
+pub fn build_v5_conn_pros_by_user_information(
+    props: Properties,
+    username: String,
+    password: String,
+    ws: bool,
+    ssl: bool,
+) -> ConnectOptions {
+    let mut conn_opts = if ws {
+        ConnectOptionsBuilder::new_ws_v5()
+    } else {
+        ConnectOptionsBuilder::new_v5()
+    };
+    if ssl {
+        let ssl_opts = SslOptionsBuilder::new()
+            .trust_store(format!(
+                "{}/../../config/example/certs/ca.pem",
+                env!("CARGO_MANIFEST_DIR")
+            ))
+            .unwrap()
+            .verify(false)
+            .disable_default_trust_store(false)
+            .finalize();
+        conn_opts.ssl_options(ssl_opts);
+    }
+    conn_opts
+        .keep_alive_interval(Duration::from_secs(600))
+        .clean_start(true)
+        .connect_timeout(Duration::from_secs(60))
+        .properties(props.clone())
+        .user_name(username)
+        .password(password)
+        .finalize()
+}
+
 #[allow(dead_code)]
 pub fn build_v3_conn_pros(mqtt_version: u32, err_pwd: bool, ws: bool, ssl: bool) -> ConnectOptions {
     let pwd = if err_pwd { err_password() } else { password() };
@@ -142,6 +181,40 @@ pub fn build_v3_conn_pros(mqtt_version: u32, err_pwd: bool, ws: bool, ssl: bool)
         .finalize()
 }
 
+#[allow(dead_code)]
+pub fn build_v3_conn_pros_by_user_information(
+    mqtt_version: u32,
+    username: String,
+    password: String,
+    ws: bool,
+    ssl: bool,
+) -> ConnectOptions {
+    let mut conn_opts = if ws {
+        ConnectOptionsBuilder::new_ws()
+    } else {
+        ConnectOptionsBuilder::with_mqtt_version(mqtt_version)
+    };
+    if ssl {
+        let ssl_opts = SslOptionsBuilder::new()
+            .trust_store(format!(
+                "{}/../../config/example/certs/ca.pem",
+                env!("CARGO_MANIFEST_DIR")
+            ))
+            .unwrap()
+            .verify(false)
+            .disable_default_trust_store(false)
+            .finalize();
+        conn_opts.ssl_options(ssl_opts);
+    }
+    conn_opts
+        .keep_alive_interval(Duration::from_secs(600))
+        .clean_session(true)
+        .connect_timeout(Duration::from_secs(50))
+        .user_name(username)
+        .password(password)
+        .finalize()
+}
+
 #[allow(dead_code)]
 pub fn build_create_pros(client_id: &str, addr: &str) -> CreateOptions {
     if client_id.is_empty() {
diff --git a/src/mqtt-broker/tests/permission34_test.rs b/src/mqtt-broker/tests/permission34_test.rs
new file mode 100644
index 000000000..e5ef9eb5a
--- /dev/null
+++ b/src/mqtt-broker/tests/permission34_test.rs
@@ -0,0 +1,233 @@
+// 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.
+
+mod common;
+// extern crate mqtt_broker;
+
+#[cfg(test)]
+mod tests {
+    use std::process;
+    use std::sync::Arc;
+
+    use common_base::tools::unique_id;
+    use grpc_clients::mqtt::admin::call::{mqtt_broker_create_user, mqtt_broker_delete_user};
+    use grpc_clients::pool::ClientPool;
+    use paho_mqtt::{Client, ReasonCode};
+    use protocol::broker_mqtt::broker_mqtt_admin::{CreateUserRequest, DeleteUserRequest};
+
+    use crate::common::{
+        broker_addr, broker_grpc_addr, broker_ssl_addr, broker_ws_addr, broker_wss_addr,
+        build_create_pros, build_v3_conn_pros_by_user_information, distinct_conn,
+    };
+
+    #[tokio::test]
+    async fn client3_permission_test() {
+        let mqtt_version = 3;
+        let client_id = unique_id();
+        let addr = broker_addr();
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let grpc_addr = vec![broker_grpc_addr()];
+
+        let username = "puser3".to_string();
+        let password = "permission".to_string();
+
+        //unregistered users are not allowed to create connections.
+        v3_permission_wrong_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //registered users are allowed to create connections.
+        create_user(
+            client_pool.clone(),
+            grpc_addr.clone(),
+            username.clone(),
+            password.clone(),
+        )
+        .await;
+        v3_permission_success_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //unregistered users are not allowed to create connections.
+        delete_user(client_pool.clone(), grpc_addr.clone(), username.clone()).await;
+        v3_permission_wrong_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+    }
+
+    #[tokio::test]
+    async fn client4_permission_test() {
+        let mqtt_version = 4;
+        let client_id = unique_id();
+        let addr = broker_addr();
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let grpc_addr = vec![broker_grpc_addr()];
+
+        let username = "puser4".to_string();
+        let password = "permission".to_string();
+
+        //unregistered users are not allowed to create connections.
+        v3_permission_wrong_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //registered users are allowed to create connections.
+        create_user(
+            client_pool.clone(),
+            grpc_addr.clone(),
+            username.clone(),
+            password.clone(),
+        )
+        .await;
+        v3_permission_success_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //unregistered users are not allowed to create connections.
+        delete_user(client_pool.clone(), grpc_addr.clone(), username.clone()).await;
+        v3_permission_wrong_test(
+            mqtt_version,
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+    }
+
+    fn v3_permission_wrong_test(
+        mqtt_version: u32,
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let create_opts = build_create_pros(client_id, addr);
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts =
+            build_v3_conn_pros_by_user_information(mqtt_version, username, password, ws, ssl);
+
+        let err = cli.connect(conn_opts).unwrap_err();
+        println!("Unable to connect:\n\t{:?}", err);
+    }
+
+    fn v3_permission_success_test(
+        mqtt_version: u32,
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let create_opts = build_create_pros(client_id, addr);
+        let cli = Client::new(create_opts).unwrap();
+
+        let conn_opts = build_v3_conn_pros_by_user_information(
+            mqtt_version,
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+        let response = cli.connect(conn_opts).unwrap();
+        let resp = response.connect_response().unwrap();
+        if ws {
+            if ssl {
+                assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+            } else {
+                assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+            }
+            assert_eq!(4, resp.mqtt_version);
+        } else {
+            if ssl {
+                assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+            } else {
+                assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+            }
+            assert_eq!(mqtt_version, resp.mqtt_version);
+        }
+        assert!(resp.session_present);
+        assert_eq!(response.reason_code(), ReasonCode::Success);
+        distinct_conn(cli);
+    }
+
+    async fn create_user(
+        client_pool: Arc<ClientPool>,
+        addrs: Vec<String>,
+        username: String,
+        password: String,
+    ) {
+        let user = CreateUserRequest {
+            username,
+            password,
+            is_superuser: false,
+        };
+        match mqtt_broker_create_user(client_pool.clone(), addrs.clone(), user.clone()).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+
+    async fn delete_user(client_pool: Arc<ClientPool>, addrs: Vec<String>, username: String) {
+        let user = DeleteUserRequest { username };
+        match mqtt_broker_delete_user(client_pool.clone(), addrs.clone(), user.clone()).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+}
diff --git a/src/mqtt-broker/tests/permission5_test.rs b/src/mqtt-broker/tests/permission5_test.rs
new file mode 100644
index 000000000..65a6f8dca
--- /dev/null
+++ b/src/mqtt-broker/tests/permission5_test.rs
@@ -0,0 +1,493 @@
+// 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.
+
+mod common;
+
+#[cfg(test)]
+mod tests {
+    use std::process;
+    use std::sync::Arc;
+
+    use common_base::tools::unique_id;
+    use grpc_clients::mqtt::admin::call::{mqtt_broker_create_user, mqtt_broker_delete_user};
+    use grpc_clients::pool::ClientPool;
+    use mqtt_broker::handler::connection::REQUEST_RESPONSE_PREFIX_NAME;
+    use paho_mqtt::{Client, PropertyCode, ReasonCode};
+    use protocol::broker_mqtt::broker_mqtt_admin::{CreateUserRequest, DeleteUserRequest};
+
+    use crate::common::{
+        broker_addr, broker_grpc_addr, broker_ssl_addr, broker_ws_addr, broker_wss_addr,
+        build_create_pros, build_v5_conn_pros_by_user_information, build_v5_pros, distinct_conn,
+    };
+
+    #[tokio::test]
+    async fn client5_permission_test() {
+        let client_id = unique_id();
+        let addr = broker_addr();
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(3));
+        let grpc_addr = vec![broker_grpc_addr()];
+
+        let username = "puser5".to_string();
+        let password = "permission".to_string();
+
+        //unregistered users are not allowed to create connections.
+        v5_permission_wrong_test(
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //registered users are allowed to create connections.
+        create_user(
+            client_pool.clone(),
+            grpc_addr.clone(),
+            username.clone(),
+            password.clone(),
+        )
+        .await;
+        v5_permission_success_test(
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+        v5_response_test(
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+        v5_assigned_client_id_test(&addr, username.clone(), password.clone(), false, false);
+        v5_request_response_test(
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+
+        //unregistered users are not allowed to create connections.
+        delete_user(client_pool.clone(), grpc_addr.clone(), username.clone()).await;
+        v5_permission_wrong_test(
+            &client_id,
+            &addr,
+            username.clone(),
+            password.clone(),
+            false,
+            false,
+        );
+    }
+
+    fn v5_permission_wrong_test(
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let create_opts = build_create_pros(client_id, addr);
+
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let props = build_v5_pros();
+        let conn_opts = build_v5_conn_pros_by_user_information(
+            props,
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+        let err = cli.connect(conn_opts).unwrap_err();
+        println!("Unable to connect:\n\t{:?}", err);
+    }
+
+    fn v5_permission_success_test(
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let mqtt_version = 5;
+        let props = build_v5_pros();
+
+        let create_opts = build_create_pros(client_id, addr);
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_v5_conn_pros_by_user_information(
+            props.clone(),
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+        match cli.connect(conn_opts) {
+            Ok(response) => {
+                let resp = response.connect_response().unwrap();
+                if ws {
+                    if ssl {
+                        assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+                    } else {
+                        assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+                    }
+                } else if ssl {
+                    assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+                } else {
+                    assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+                }
+                assert_eq!(mqtt_version, resp.mqtt_version);
+                // assert!(!resp.session_present);
+                assert_eq!(response.reason_code(), ReasonCode::Success);
+            }
+            Err(e) => {
+                println!("Unable to connect:\n\t{:?}", e);
+                process::exit(1);
+            }
+        }
+        distinct_conn(cli);
+    }
+
+    fn v5_assigned_client_id_test(
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let mqtt_version = 5;
+        let client_id = "".to_string();
+        let props = build_v5_pros();
+
+        let create_opts = build_create_pros(&client_id, addr);
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_v5_conn_pros_by_user_information(
+            props.clone(),
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+        match cli.connect(conn_opts) {
+            Ok(response) => {
+                let resp = response.connect_response().unwrap();
+                if ws {
+                    if ssl {
+                        assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+                    } else {
+                        assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+                    }
+                } else if ssl {
+                    assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+                } else {
+                    assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+                }
+                assert_eq!(mqtt_version, resp.mqtt_version);
+                // assert!(resp.session_present);
+                assert_eq!(response.reason_code(), ReasonCode::Success);
+
+                let resp_pros = response.properties();
+                let assign_client_id = resp_pros
+                    .get(PropertyCode::AssignedClientIdentifer)
+                    .unwrap()
+                    .get_string()
+                    .unwrap();
+                assert!(!assign_client_id.is_empty());
+                assert_eq!(assign_client_id.len(), unique_id().len());
+            }
+            Err(e) => {
+                println!("Unable to connect:\n\t{:?}", e);
+                process::exit(1);
+            }
+        }
+        distinct_conn(cli);
+    }
+
+    fn v5_request_response_test(
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let mqtt_version = 5;
+
+        let pros = build_v5_pros();
+
+        let create_opts = build_create_pros(client_id, addr);
+
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_v5_conn_pros_by_user_information(
+            pros.clone(),
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+
+        match cli.connect(conn_opts) {
+            Ok(response) => {
+                let resp = response.connect_response().unwrap();
+                // response
+                if ws {
+                    if ssl {
+                        assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+                    } else {
+                        assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+                    }
+                } else if ssl {
+                    assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+                } else {
+                    assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+                }
+                assert_eq!(mqtt_version, resp.mqtt_version);
+                assert_eq!(response.reason_code(), ReasonCode::Success);
+
+                // properties
+                let resp_pros = response.properties();
+                assert!(resp_pros
+                    .get_string(PropertyCode::ResponseInformation)
+                    .is_none());
+            }
+            Err(e) => {
+                println!("Unable to connect:\n\t{:?}", e);
+                process::exit(1);
+            }
+        }
+        distinct_conn(cli);
+    }
+
+    fn v5_response_test(
+        client_id: &str,
+        addr: &str,
+        username: String,
+        password: String,
+        ws: bool,
+        ssl: bool,
+    ) {
+        let mqtt_version = 5;
+
+        let mut pros = build_v5_pros();
+        pros.push_val(PropertyCode::RequestResponseInformation, 1)
+            .unwrap();
+
+        let create_opts = build_create_pros(client_id, addr);
+
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_v5_conn_pros_by_user_information(
+            pros.clone(),
+            username.clone(),
+            password.clone(),
+            ws,
+            ssl,
+        );
+
+        match cli.connect(conn_opts) {
+            Ok(response) => {
+                let resp = response.connect_response().unwrap();
+                // response
+                if ws {
+                    if ssl {
+                        assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+                    } else {
+                        assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+                    }
+                } else if ssl {
+                    assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+                } else {
+                    assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+                }
+                assert_eq!(mqtt_version, resp.mqtt_version);
+                assert_eq!(response.reason_code(), ReasonCode::Success);
+
+                // properties
+                let resp_pros = response.properties();
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::SessionExpiryInterval)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    3
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::ReceiveMaximum)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    65535
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::MaximumQos)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    2
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::RetainAvailable)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    1
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::MaximumPacketSize)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    10485760
+                );
+
+                assert!(resp_pros
+                    .get(PropertyCode::AssignedClientIdentifer)
+                    .is_none());
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::TopicAliasMaximum)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    65535
+                );
+
+                assert!(resp_pros.get(PropertyCode::ReasonString).is_none());
+
+                assert!(resp_pros.get(PropertyCode::UserProperty).is_none());
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::WildcardSubscriptionAvailable)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    1
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::SubscriptionIdentifiersAvailable)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    1
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::SharedSubscriptionAvailable)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    1
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::ServerKeepAlive)
+                        .unwrap()
+                        .get_int()
+                        .unwrap(),
+                    1200
+                );
+
+                assert_eq!(
+                    resp_pros
+                        .get(PropertyCode::ResponseInformation)
+                        .unwrap()
+                        .get_string()
+                        .unwrap(),
+                    REQUEST_RESPONSE_PREFIX_NAME.to_string()
+                );
+
+                assert!(resp_pros.get(PropertyCode::ServerReference).is_none());
+
+                assert!(resp_pros.get(PropertyCode::AuthenticationMethod).is_none());
+                assert!(resp_pros.get(PropertyCode::AuthenticationData).is_none());
+            }
+            Err(e) => {
+                println!("Unable to connect:\n\t{:?}", e);
+                process::exit(1);
+            }
+        }
+        distinct_conn(cli);
+    }
+
+    async fn create_user(
+        client_pool: Arc<ClientPool>,
+        addrs: Vec<String>,
+        username: String,
+        password: String,
+    ) {
+        let user = CreateUserRequest {
+            username,
+            password,
+            is_superuser: false,
+        };
+        match mqtt_broker_create_user(client_pool.clone(), addrs.clone(), user.clone()).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+
+    async fn delete_user(client_pool: Arc<ClientPool>, addrs: Vec<String>, username: String) {
+        let user = DeleteUserRequest { username };
+        match mqtt_broker_delete_user(client_pool.clone(), addrs.clone(), user.clone()).await {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+}