[CIT-270] configure moquitto port (#248)

* [CIT-270] Configure mosquitto port

* [CIT-270] added and updated tests

* [CIT-270] Added document for configuring port

* cargo fmt

* Fixed markup issues

* Updated the configure mqtt port doc link

* Added configure mqtt port doc link

* fixed merge conflicts

* [CIT-270] rename mosquitto to mqtt

* [CIT-270] fixed review comments

* [CIT-270] fix the tests

* [CIT-270] use mqtt.port in mappers

* [CIT-270] use mqtt.port in dm agent

* [CIT-270] update the doc

* [CIT-270] addressed review comments

* [CIT-270] renamed document and fixed comments

Co-authored-by: pradeekumar.kj@softwareag.com <pradeekumar.kj@softwareag.com>
Co-authored-by: Pradeep K J <Pradeep K J pradeekumar.kj@softwareag.com>

7 files changed, 157 insertions, 4 deletions

diff --git a/tedge_config/src/models/mod.rs b/tedge_config/src/models/mod.rs
index 9ea27a70..e74f756c 100644
--- a/tedge_config/src/models/mod.rs
+++ b/tedge_config/src/models/mod.rs
@@ -1,5 +1,5 @@
 pub mod connect_url;
 pub mod file_path;
 pub mod flag;
-
-pub use self::{connect_url::*, file_path::*, flag::*};
+pub mod port;
+pub use self::{connect_url::*, file_path::*, flag::*, port::*};
diff --git a/tedge_config/src/models/port.rs b/tedge_config/src/models/port.rs
new file mode 100644
index 00000000..f6cfef9c
--- /dev/null
+++ b/tedge_config/src/models/port.rs
@@ -0,0 +1,56 @@
+use std::convert::{TryFrom, TryInto};
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub struct Port(pub u16);
+
+#[derive(thiserror::Error, Debug)]
+#[error("Invalid port number: '{input}'.")]
+pub struct InvalidPortNumber {
+    input: String,
+}
+
+impl TryFrom<String> for Port {
+    type Error = InvalidPortNumber;
+
+    fn try_from(input: String) -> Result<Self, Self::Error> {
+        input
+            .as_str()
+            .parse::<u16>()
+            .map_err(|_| InvalidPortNumber { input })
+            .map(Port)
+    }
+}
+
+impl TryInto<String> for Port {
+    type Error = std::convert::Infallible;
+
+    fn try_into(self) -> Result<String, Self::Error> {
+        Ok(format!("{}", self.0))
+    }
+}
+
+impl Into<u16> for Port {
+    fn into(self) -> u16 {
+        self.0
+    }
+}
+
+#[cfg(test)]
+use assert_matches::*;
+#[test]
+fn conversion_from_valid_port_succeeds() {
+    assert_matches!(Port::try_from("1234".to_string()), Ok(Port(1234)));
+}
+
+#[test]
+fn conversion_from_longer_integer_fails() {
+    assert_matches!(
+        Port::try_from("66000".to_string()),
+        Err(InvalidPortNumber { .. })
+    );
+}
+
+#[test]
+fn conversion_from_port_to_string() {
+    assert_matches!(TryInto::<String>::try_into(Port(1234)), Ok(port_str) if port_str == "1234");
+}
diff --git a/tedge_config/src/settings.rs b/tedge_config/src/settings.rs
index 35092393..353db4c1 100644
--- a/tedge_config/src/settings.rs
+++ b/tedge_config/src/settings.rs
@@ -126,6 +126,20 @@ impl ConfigSetting for AzureRootCertPathSetting {
     type Value = FilePath;
 }
 
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub struct MqttPortSetting;
+
+impl ConfigSetting for MqttPortSetting {
+    const KEY: &'static str = "mqtt.port";
+
+    const DESCRIPTION: &'static str = concat!(
+        "Mqtt broker port, which is used by the mqtt clients to publish or subscribe. ",
+        "Example: listener 1883"
+    );
+
+    type Value = Port;
+}
+
 ///
 /// Boolean whether Azure mapper should add timestamp if timestamp is not added in the incoming payload.
 ///
diff --git a/tedge_config/src/tedge_config.rs b/tedge_config/src/tedge_config.rs
index d4d2fbbd..bacc20c8 100644
--- a/tedge_config/src/tedge_config.rs
+++ b/tedge_config/src/tedge_config.rs
@@ -223,6 +223,27 @@ impl ConfigSettingAccessor<C8yRootCertPathSetting> for TEdgeConfig {
     }
 }
 
+impl ConfigSettingAccessor<MqttPortSetting> for TEdgeConfig {
+    fn query(&self, _setting: MqttPortSetting) -> ConfigSettingResult<Port> {
+        Ok(self
+            .data
+            .mqtt
+            .port
+            .map(Port)
+            .unwrap_or_else(|| self.config_defaults.default_mqtt_port))
+    }
+
+    fn update(&mut self, _setting: MqttPortSetting, value: Port) -> ConfigSettingResult<()> {
+        self.data.mqtt.port = Some(value.into());
+        Ok(())
+    }
+
+    fn unset(&mut self, _setting: MqttPortSetting) -> ConfigSettingResult<()> {
+        self.data.mqtt.port = None;
+        Ok(())
+    }
+}
+
 /// Generic extension trait implementation for all `ConfigSetting`s of `TEdgeConfig`
 /// that provide `TryFrom`/`TryInto` implementations for `String`.
 impl<T, E, F> ConfigSettingAccessorStringExt<T> for TEdgeConfig
diff --git a/tedge_config/src/tedge_config_defaults.rs b/tedge_config/src/tedge_config_defaults.rs
index 6896c51b..3e5764fb 100644
--- a/tedge_config/src/tedge_config_defaults.rs
+++ b/tedge_config/src/tedge_config_defaults.rs
@@ -1,8 +1,10 @@
 use crate::models::FilePath;
+use crate::Port;
 use crate::TEdgeConfigLocation;
 use std::path::Path;
 
 const DEFAULT_ETC_PATH: &str = "/etc";
+const DEFAULT_PORT: u16 = 1883;
 
 /// Stores default values for use by `TEdgeConfig` in case no configuration setting
 /// is available.
@@ -29,6 +31,9 @@ pub struct TEdgeConfigDefaults {
 
     /// Default path for c8y root certificates
     pub default_c8y_root_cert_path: FilePath,
+
+    /// Default port for mqtt
+    pub default_mqtt_port: Port,
 }
 
 impl From<&TEdgeConfigLocation> for TEdgeConfigDefaults {
@@ -47,6 +52,7 @@ impl From<&TEdgeConfigLocation> for TEdgeConfigDefaults {
                 .into(),
             default_azure_root_cert_path: system_cert_path.clone().into(),
             default_c8y_root_cert_path: system_cert_path.into(),
+            default_mqtt_port: Port(DEFAULT_PORT),
         }
     }
 }
@@ -66,7 +72,8 @@ fn test_from_tedge_config_location() {
                 "/opt/etc/_tedge/device-certs/tedge-private-key.pem"
             ),
             default_azure_root_cert_path: FilePath::from("/etc/ssl/certs"),
-            default_c8y_root_cert_path: FilePath::from("/etc/ssl/certs")
+            default_c8y_root_cert_path: FilePath::from("/etc/ssl/certs"),
+            default_mqtt_port: Port(DEFAULT_PORT),
         }
     );
 }
diff --git a/tedge_config/src/tedge_config_dto.rs b/tedge_config/src/tedge_config_dto.rs
index 0d00ffbe..0b1d9c19 100644
--- a/tedge_config/src/tedge_config_dto.rs
+++ b/tedge_config/src/tedge_config_dto.rs
@@ -13,8 +13,12 @@ pub(crate) struct TEdgeConfigDto {
     /// Captures the configurations required to connect to Cumulocity
     #[serde(default)]
     pub(crate) c8y: CumulocityConfigDto,
+
     #[serde(default)]
     pub(crate) azure: AzureConfigDto,
+
+    #[serde(default)]
+    pub(crate) mqtt: MqttConfigDto,
 }
 
 /// Represents the device specific configurations defined in the [device] section
@@ -63,3 +67,9 @@ pub(crate) struct AzureConfigDto {
     pub(crate) root_cert_path: Option<FilePath>,
     pub(crate) mapper_timestamp: Option<bool>,
 }
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct MqttConfigDto {
+    pub(crate) port: Option<u16>,
+}
diff --git a/tedge_config/tests/test_tedge_config.rs b/tedge_config/tests/test_tedge_config.rs
index 8bf404c4..7c82965c 100644
--- a/tedge_config/tests/test_tedge_config.rs
+++ b/tedge_config/tests/test_tedge_config.rs
@@ -20,6 +20,9 @@ connect = "true"
 url = "MyAzure.azure-devices.net"
 root_cert_path = "/path/to/azure/root/cert"
 connect = "false"
+
+[mqtt]
+port = 1234
 "#;
 
     let (_tempdir, config_location) = create_temp_tedge_config(toml_conf)?;
@@ -56,6 +59,8 @@ connect = "false"
         FilePath::from("/path/to/azure/root/cert")
     );
 
+    assert_eq!(config.query(MqttPortSetting)?, Port(1234));
+
     Ok(())
 }
 
@@ -73,6 +78,9 @@ root_cert_path = "/path/to/c8y/root/cert"
 [azure]
 url = "MyAzure.azure-devices.net"
 root_cert_path = "/path/to/azure/root/cert"
+
+[mqtt]
+port = 1883
 "#;
 
     let (_tempdir, config_location) = create_temp_tedge_config(toml_conf)?;
@@ -86,6 +94,7 @@ root_cert_path = "/path/to/azure/root/cert"
 
     let updated_c8y_url = "other-tenant.cumulocity.com";
     let updated_azure_url = "OtherAzure.azure-devices.net";
+    let updated_mqtt_port = Port(2345);
 
     {
         let mut config = config_repo.load()?;
@@ -120,6 +129,7 @@ root_cert_path = "/path/to/azure/root/cert"
         config.update(C8yUrlSetting, ConnectUrl::try_from(updated_c8y_url)?)?;
         config.unset(C8yRootCertPathSetting)?;
         config.update(AzureUrlSetting, ConnectUrl::try_from(updated_azure_url)?)?;
+        config.update(MqttPortSetting, updated_mqtt_port)?;
         config.unset(AzureRootCertPathSetting)?;
         config_repo.store(config)?;
     }
@@ -148,6 +158,8 @@ root_cert_path = "/path/to/azure/root/cert"
             config.query(AzureRootCertPathSetting)?,
             FilePath::from("default_azure_root_cert_path")
         );
+
+        assert_eq!(config.query(MqttPortSetting)?, updated_mqtt_port);
     }
 
     Ok(())
@@ -297,6 +309,8 @@ fn test_parse_config_empty_file() -> Result<(), TEdgeConfigError> {
         config.query(AzureRootCertPathSetting)?,
         FilePath::from("/etc/ssl/certs")
     );
+
+    assert_eq!(config.query(MqttPortSetting)?, Port(1883));
     Ok(())
 }
 
@@ -330,6 +344,28 @@ hello="tedge"
 }
 
 #[test]
+fn test_invalid_mqtt_port() -> Result<(), TEdgeConfigError> {
+    let toml_conf = r#"
+[mqtt]
+port = "1883"
+"#;
+
+    let (_tempdir, config_location) = create_temp_tedge_config(toml_conf)?;
+    let result = TEdgeConfigRepository::new(config_location).load();
+
+    let expected_err =
+        "invalid type: string \"1883\", expected u16 for key `mqtt.port` at line 3 column 8";
+
+    match result {
+        Err(TEdgeConfigError::TOMLParseError(err)) => assert_eq!(err.to_string(), expected_err),
+
+        _ => assert!(false, "Expected the parsing to fail with TOMLParseError"),
+    }
+
+    Ok(())
+}
+
+#[test]
 fn test_parse_invalid_toml_file() -> Result<(), TEdgeConfigError> {
     let toml_conf = r#"
         <abcde>
@@ -360,6 +396,9 @@ root_cert_path = "/path/to/c8y/root/cert"
 [azure]
 url = "MyAzure.azure-devices.net"
 root_cert_path = "/path/to/azure/root/cert"
+
+[mqtt]
+port = 1024
 "#;
 
     let (_tempdir, config_location) = create_temp_tedge_config(toml_conf)?;
@@ -397,6 +436,9 @@ root_cert_path = "/path/to/azure/root/cert"
 
     config.update(C8yUrlSetting, updated_c8y_url.clone())?;
 
+    let updated_mqtt_port = Port(2048);
+    config.update(MqttPortSetting, updated_mqtt_port.clone())?;
+
     config.unset(C8yRootCertPathSetting)?;
 
     assert_eq!(
@@ -413,6 +455,8 @@ root_cert_path = "/path/to/azure/root/cert"
         config.query(C8yRootCertPathSetting)?,
         FilePath::from("/etc/ssl/certs")
     );
+
+    assert_eq!(config.query(MqttPortSetting)?, updated_mqtt_port);
     Ok(())
 }
 
@@ -529,7 +573,7 @@ cert_path = "/path/to/cert"
 }
 
 #[test]
-fn test_device_id_is_extraxted_from_device_certificate() -> Result<(), TEdgeConfigError> {
+fn test_device_id_is_extracted_from_device_certificate() -> Result<(), TEdgeConfigError> {
     let toml_conf = r#"
 [device]
 cert_path = "/path/to/cert"
@@ -564,6 +608,7 @@ fn dummy_tedge_config_defaults() -> TEdgeConfigDefaults {
         default_device_key_path: FilePath::from("/dev/null"),
         default_c8y_root_cert_path: FilePath::from("/dev/null"),
         default_azure_root_cert_path: FilePath::from("/dev/null"),
+        default_mqtt_port: Port(1883),
     }
 }