Move mqtt_server to separate crate such that it is usable by other crates tests (#372)

Signed-off-by: Lukasz Woznicki <lukasz.woznicki@softwareag.com>

7 files changed, 24 insertions, 15 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 2c67b0a4..4f7ca7bc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1397,6 +1397,7 @@ dependencies = [
  "rumqttd",
  "rumqttlog",
  "serde",
+ "tedge_utils",
  "thiserror",
  "tokio",
  "tokio-test",
@@ -2599,7 +2600,11 @@ dependencies = [
 name = "tedge_utils"
 version = "0.2.3"
 dependencies = [
+ "anyhow",
  "assert_matches",
+ "futures",
+ "rumqttd",
+ "rumqttlog",
  "tempfile",
  "thiserror",
  "tokio",
diff --git a/common/mqtt_client/Cargo.toml b/common/mqtt_client/Cargo.toml
index 93955691..2b62cf37 100644
--- a/common/mqtt_client/Cargo.toml
+++ b/common/mqtt_client/Cargo.toml
@@ -14,7 +14,7 @@ mockall = "0.10"
 async-trait = "0.1"
 
 [dev-dependencies]
-anyhow="1.0"
+anyhow= "1.0"
 async-log = "2.0"
 env_logger = "0.9"
 futures = "0.3"
@@ -23,8 +23,9 @@ json = "0.12"
 log = "0.4"
 rand = "0.8"
 rumqttd = "0.7"
-serde = "1.0.126"
 rumqttlog = "0.7"
+serde = "1.0.126"
+tedge_utils = { path = "../../common/tedge_utils" }
 tokio-test = "0.4"
 
 [features]
diff --git a/common/mqtt_client/tests/mqtt_pub_sub_test.rs b/common/mqtt_client/tests/mqtt_pub_sub_test.rs
index 71beb205..3f523911 100644
--- a/common/mqtt_client/tests/mqtt_pub_sub_test.rs
+++ b/common/mqtt_client/tests/mqtt_pub_sub_test.rs
@@ -1,6 +1,6 @@
-mod rumqttd_broker;
 use mqtt_client::{Client, Message, MqttClient, Topic, TopicFilter};
 use std::time::Duration;
+use tedge_utils::test_mqtt_server::start_broker_local;
 use tokio::time::sleep;
 
 const MQTTTESTPORT: u16 = 58586;
@@ -8,8 +8,7 @@ const MQTTTESTPORT: u16 = 58586;
 #[test]
 fn sending_and_receiving_a_message() {
     async fn scenario(payload: String) -> Result<Option<Message>, mqtt_client::MqttClientError> {
-        let _mqtt_server_handle =
-            tokio::spawn(async { rumqttd_broker::start_broker_local(MQTTTESTPORT).await });
+        let _mqtt_server_handle = tokio::spawn(async { start_broker_local(MQTTTESTPORT).await });
         let topic = Topic::new("test/uubpb9wyi9asi46l624f")?;
         let subscriber = Client::connect(
             "subscribe",
@@ -44,8 +43,7 @@ fn sending_and_receiving_a_message() {
 async fn subscribing_to_many_topics() -> Result<(), anyhow::Error> {
     // Given an MQTT broker
     let mqtt_port: u16 = 55555;
-    let _mqtt_server_handle =
-        tokio::spawn(async move { rumqttd_broker::start_broker_local(mqtt_port).await });
+    let _mqtt_server_handle = tokio::spawn(async move { start_broker_local(mqtt_port).await });
 
     // And an MQTT client connected to that server
     let subscriber = Client::connect(
diff --git a/common/mqtt_client/tests/packet_size_tests.rs b/common/mqtt_client/tests/packet_size_tests.rs
index 113bb523..834cd48a 100644
--- a/common/mqtt_client/tests/packet_size_tests.rs
+++ b/common/mqtt_client/tests/packet_size_tests.rs
@@ -1,8 +1,7 @@
+use futures::future::TryFutureExt;
 use mqtt_client::{Client, Message, MqttClient, MqttClientError, QoS, Topic, TopicFilter};
 use rumqttc::StateError;
-mod rumqttd_broker;
-use futures::future::TryFutureExt;
-
+use tedge_utils::test_mqtt_server::start_broker_local;
 use tokio::time::Duration;
 
 const MQTTTESTPORT1: u16 = 58584;
@@ -18,8 +17,7 @@ enum TestJoinError {
 // This checks the mqtt packets are within the limit or not
 async fn packet_size_within_limit() -> Result<(), anyhow::Error> {
     // Start the local broker
-    let _mqtt_server_handle =
-        tokio::spawn(async { rumqttd_broker::start_broker_local(MQTTTESTPORT1).await });
+    let _mqtt_server_handle = tokio::spawn(async { start_broker_local(MQTTTESTPORT1).await });
     // Start the subscriber
     let subscriber = tokio::spawn(async move { subscribe_until_3_messages_received().await });
 
@@ -43,8 +41,7 @@ async fn packet_size_within_limit() -> Result<(), anyhow::Error> {
 // This checks the mqtt packet size that exceeds the limit
 async fn packet_size_exceeds_limit() -> Result<(), anyhow::Error> {
     // Start the broker
-    let _mqtt_server_handle =
-        tokio::spawn(async { rumqttd_broker::start_broker_local(MQTTTESTPORT2).await });
+    let _mqtt_server_handle = tokio::spawn(async { start_broker_local(MQTTTESTPORT2).await });
 
     // Start the publisher and publish a message
     let publish = tokio::spawn(async { publish_big_message_wait_for_error().await });
diff --git a/common/tedge_utils/Cargo.toml b/common/tedge_utils/Cargo.toml
index 4f3bc3a5..ed0948db 100644
--- a/common/tedge_utils/Cargo.toml
+++ b/common/tedge_utils/Cargo.toml
@@ -7,9 +7,16 @@ license = "Apache-2.0"
 description = "tedge_utils provide utilities for thin-edge.io components"
 
 [dependencies]
+anyhow = "1.0"
+futures = "0.3"
+rumqttd = "0.7"
+rumqttlog = "0.7"
 tempfile = "3.2"
 thiserror = "1.0"
-tokio = { version = "1.8", default_features = false, features = ["signal"] }
+tokio = { version = "1.9", default_features = false, features = ["signal", "sync", "macros"] }
+
 
 [dev-dependencies]
 assert_matches = "1.5"
+
+
diff --git a/common/tedge_utils/src/lib.rs b/common/tedge_utils/src/lib.rs
index 9139bbc9..796f3965 100644
--- a/common/tedge_utils/src/lib.rs
+++ b/common/tedge_utils/src/lib.rs
@@ -1,3 +1,4 @@
 pub mod fs;
 pub mod paths;
 pub mod signals;
+pub mod test_mqtt_server;
diff --git a/common/mqtt_client/tests/rumqttd_broker.rs b/common/tedge_utils/src/test_mqtt_server.rs
index ec377741..ec377741 100644
--- a/common/mqtt_client/tests/rumqttd_broker.rs
+++ b/common/tedge_utils/src/test_mqtt_server.rs