Feature/599/restart device cloud operation (#662)

* cloud restart operation #599

Signed-off-by: initard <solo@softwareag.com>

* operation set to executing c8y #599

Signed-off-by: initard <solo@softwareag.com>

* cargo fmt

Signed-off-by: initard <solo@softwareag.com>

* removing debug hard-coded echo

Signed-off-by: initard <solo@softwareag.com>

* updated handle_restart_opeartion signature in test #599

Signed-off-by: initard <solo@softwareag.com>

* test update for c8y_Restart

Signed-off-by: initard <solo@softwareag.com>

* removed panic on empty restart request #599

Signed-off-by: initard <solo@softwareag.com>

Co-authored-by: initard <solo@softwareag.com>

3 files changed, 70 insertions, 7 deletions

diff --git a/crates/core/tedge_mapper/src/sm_c8y_mapper/mapper.rs b/crates/core/tedge_mapper/src/sm_c8y_mapper/mapper.rs
index f40b1873..e838c508 100644
--- a/crates/core/tedge_mapper/src/sm_c8y_mapper/mapper.rs
+++ b/crates/core/tedge_mapper/src/sm_c8y_mapper/mapper.rs
@@ -3,7 +3,7 @@ use crate::sm_c8y_mapper::json_c8y::{C8yCreateEvent, C8yManagedObject};
 use crate::sm_c8y_mapper::{error::*, json_c8y::C8yUpdateSoftwareListResponse, topic::*};
 use crate::{component::TEdgeComponent, sm_c8y_mapper::json_c8y::InternalIdResponse};
 use async_trait::async_trait;
-use c8y_smartrest::smartrest_deserializer::SmartRestLogRequest;
+use c8y_smartrest::smartrest_deserializer::{SmartRestLogRequest, SmartRestRestartRequest};
 use c8y_smartrest::smartrest_serializer::CumulocitySupportedOperations;
 use c8y_smartrest::{
     error::SmartRestDeserializerError,
@@ -16,8 +16,8 @@ use c8y_smartrest::{
 };
 use chrono::{DateTime, FixedOffset, Local};
 use json_sm::{
-    Auth, DownloadInfo, Jsonify, OperationStatus, SoftwareListRequest, SoftwareListResponse,
-    SoftwareUpdateResponse,
+    Auth, DownloadInfo, Jsonify, OperationStatus, RestartOperationRequest,
+    RestartOperationResponse, SoftwareListRequest, SoftwareListResponse, SoftwareUpdateResponse,
 };
 use mqtt_client::{Client, MqttClient, MqttClientError, MqttMessageStream, Topic, TopicFilter};
 use reqwest::Url;
@@ -77,6 +77,7 @@ impl CumulocitySoftwareManagement {
         let mut topic_filter = TopicFilter::new(IncomingTopic::SoftwareListResponse.as_str())?;
         topic_filter.add(IncomingTopic::SoftwareUpdateResponse.as_str())?;
         topic_filter.add(IncomingTopic::SmartRestRequest.as_str())?;
+        topic_filter.add(IncomingTopic::RestartResponse.as_str())?;
         let messages = self.client.subscribe(topic_filter).await?;
 
         Ok(messages)
@@ -140,6 +141,9 @@ impl CumulocitySoftwareManagement {
             "522" => {
                 let () = self.forward_log_request(payload).await?;
             }
+            "510" => {
+                let () = self.forward_restart_request(payload).await?;
+            }
             _ => {
                 return Err(SMCumulocityMapperError::InvalidMqttMessage);
             }
@@ -171,6 +175,11 @@ impl CumulocitySoftwareManagement {
                         .publish_operation_status(message.payload_str()?)
                         .await?;
                 }
+                IncomingTopic::RestartResponse => {
+                    let () = self
+                        .publish_restart_operation_status(message.payload_str()?)
+                        .await?;
+                }
                 IncomingTopic::SmartRestRequest => {
                     debug!("Cumulocity");
                     let () = self.process_smartrest(message.payload_str()?).await?;
@@ -269,6 +278,41 @@ impl CumulocitySoftwareManagement {
         Ok(())
     }
 
+    async fn publish_restart_operation_status(
+        &self,
+        json_response: &str,
+    ) -> Result<(), SMCumulocityMapperError> {
+        let response = RestartOperationResponse::from_json(json_response)?;
+        let topic = OutgoingTopic::SmartRestResponse.to_topic()?;
+
+        match response.status() {
+            OperationStatus::Executing => {
+                let smartrest_set_operation = SmartRestSetOperationToExecuting::new(
+                    CumulocitySupportedOperations::C8yRestartRequest,
+                )
+                .to_smartrest()?;
+
+                let () = self.publish(&topic, smartrest_set_operation).await?;
+            }
+            OperationStatus::Successful => {
+                let smartrest_set_operation = SmartRestSetOperationToSuccessful::new(
+                    CumulocitySupportedOperations::C8yRestartRequest,
+                )
+                .to_smartrest()?;
+                let () = self.publish(&topic, smartrest_set_operation).await?;
+            }
+            OperationStatus::Failed => {
+                let smartrest_set_operation = SmartRestSetOperationToFailed::new(
+                    CumulocitySupportedOperations::C8yRestartRequest,
+                    "Restart Failed".into(),
+                )
+                .to_smartrest()?;
+                let () = self.publish(&topic, smartrest_set_operation).await?;
+            }
+        }
+        Ok(())
+    }
+
     async fn set_log_file_request_executing(&self) -> Result<(), SMCumulocityMapperError> {
         let topic = OutgoingTopic::SmartRestResponse.to_topic()?;
         let smartrest_set_operation_status =
@@ -367,6 +411,19 @@ impl CumulocitySoftwareManagement {
         Ok(())
     }
 
+    async fn forward_restart_request(
+        &self,
+        smartrest: &str,
+    ) -> Result<(), SMCumulocityMapperError> {
+        let topic = OutgoingTopic::RestartRequest.to_topic()?;
+        let _ = SmartRestRestartRequest::from_smartrest(smartrest)?;
+
+        let request = RestartOperationRequest::new();
+        let () = self.publish(&topic, request.to_json()?).await?;
+
+        Ok(())
+    }
+
     async fn publish(&self, topic: &Topic, payload: String) -> Result<(), MqttClientError> {
         let () = self
             .client
diff --git a/crates/core/tedge_mapper/src/sm_c8y_mapper/tests.rs b/crates/core/tedge_mapper/src/sm_c8y_mapper/tests.rs
index 964214ba..bd3515e8 100644
--- a/crates/core/tedge_mapper/src/sm_c8y_mapper/tests.rs
+++ b/crates/core/tedge_mapper/src/sm_c8y_mapper/tests.rs
@@ -49,7 +49,7 @@ async fn mapper_publishes_a_supported_operation_and_a_pending_operations_onto_c8
         &mut messages,
         TEST_TIMEOUT_MS,
         vec![
-            "114,c8y_SoftwareUpdate,c8y_LogfileRequest\n",
+            "114,c8y_SoftwareUpdate,c8y_LogfileRequest,c8y_Restart\n",
             "118,software-management\n",
             "500\n",
         ],
@@ -116,7 +116,7 @@ async fn mapper_publishes_software_update_status_onto_c8y_topic() {
         &mut messages,
         TEST_TIMEOUT_MS,
         vec![
-            "114,c8y_SoftwareUpdate,c8y_LogfileRequest\n",
+            "114,c8y_SoftwareUpdate,c8y_LogfileRequest,c8y_Restart\n",
             "118,software-management\n",
             "500\n",
         ],
@@ -181,7 +181,7 @@ async fn mapper_publishes_software_update_failed_status_onto_c8y_topic() {
         &mut messages,
         TEST_TIMEOUT_MS,
         vec![
-            "114,c8y_SoftwareUpdate,c8y_LogfileRequest\n",
+            "114,c8y_SoftwareUpdate,c8y_LogfileRequest,c8y_Restart\n",
             "118,software-management\n",
             "500\n",
         ],
@@ -365,7 +365,7 @@ async fn mapper_publishes_software_update_request_with_wrong_action() {
         &mut messages,
         TEST_TIMEOUT_MS,
         vec![
-            "114,c8y_SoftwareUpdate,c8y_LogfileRequest\n",
+            "114,c8y_SoftwareUpdate,c8y_LogfileRequest,c8y_Restart\n",
             "118,software-management\n",
             "500\n",
         ],
diff --git a/crates/core/tedge_mapper/src/sm_c8y_mapper/topic.rs b/crates/core/tedge_mapper/src/sm_c8y_mapper/topic.rs
index 16cbcf74..d87ee33e 100644
--- a/crates/core/tedge_mapper/src/sm_c8y_mapper/topic.rs
+++ b/crates/core/tedge_mapper/src/sm_c8y_mapper/topic.rs
@@ -7,6 +7,7 @@ pub(crate) enum IncomingTopic {
     SoftwareListResponse,
     SoftwareUpdateResponse,
     SmartRestRequest,
+    RestartResponse,
 }
 
 impl IncomingTopic {
@@ -15,6 +16,7 @@ impl IncomingTopic {
             Self::SoftwareListResponse => r#"tedge/commands/res/software/list"#,
             Self::SoftwareUpdateResponse => r#"tedge/commands/res/software/update"#,
             Self::SmartRestRequest => r#"c8y/s/ds"#,
+            Self::RestartResponse => r#"tedge/commands/res/control/restart"#,
         }
     }
 }
@@ -27,6 +29,7 @@ impl TryFrom<String> for IncomingTopic {
             r#"tedge/commands/res/software/list"# => Ok(IncomingTopic::SoftwareListResponse),
             r#"tedge/commands/res/software/update"# => Ok(IncomingTopic::SoftwareUpdateResponse),
             r#"c8y/s/ds"# => Ok(IncomingTopic::SmartRestRequest),
+            r#"tedge/commands/res/control/restart"# => Ok(IncomingTopic::RestartResponse),
             err => Err(MapperTopicError::UnknownTopic {
                 topic: err.to_string(),
             }),
@@ -55,6 +58,7 @@ pub(crate) enum OutgoingTopic {
     SoftwareListRequest,
     SoftwareUpdateRequest,
     SmartRestResponse,
+    RestartRequest,
 }
 
 impl OutgoingTopic {
@@ -63,6 +67,7 @@ impl OutgoingTopic {
             Self::SoftwareListRequest => r#"tedge/commands/req/software/list"#,
             Self::SoftwareUpdateRequest => r#"tedge/commands/req/software/update"#,
             Self::SmartRestResponse => r#"c8y/s/us"#,
+            Self::RestartRequest => r#"tedge/commands/req/control/restart"#,
         }
     }
 
@@ -71,6 +76,7 @@ impl OutgoingTopic {
             Self::SoftwareListRequest => Topic::new(Self::SoftwareListRequest.as_str()),
             Self::SoftwareUpdateRequest => Topic::new(Self::SoftwareUpdateRequest.as_str()),
             Self::SmartRestResponse => Topic::new(Self::SmartRestResponse.as_str()),
+            Self::RestartRequest => Topic::new(Self::RestartRequest.as_str()),
         }
     }
 }