diff options
Diffstat (limited to 'crates')
-rw-r--r-- | crates/core/c8y_smartrest/src/operations.rs | 2 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/c8y/converter.rs | 67 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/c8y/error.rs | 8 |
3 files changed, 57 insertions, 20 deletions
diff --git a/crates/core/c8y_smartrest/src/operations.rs b/crates/core/c8y_smartrest/src/operations.rs index c70f83b3..c7bc7c59 100644 --- a/crates/core/c8y_smartrest/src/operations.rs +++ b/crates/core/c8y_smartrest/src/operations.rs @@ -25,7 +25,7 @@ pub struct OnMessageExec { #[serde(rename_all = "lowercase")] pub struct Operation { #[serde(skip)] - name: String, + pub name: String, exec: Option<OnMessageExec>, } diff --git a/crates/core/tedge_mapper/src/c8y/converter.rs b/crates/core/tedge_mapper/src/c8y/converter.rs index bc61b78b..1ae5b4dc 100644 --- a/crates/core/tedge_mapper/src/c8y/converter.rs +++ b/crates/core/tedge_mapper/src/c8y/converter.rs @@ -288,7 +288,11 @@ async fn parse_c8y_topics( Err( ref err @ CumulocityMapperError::FromSmartRestDeserializer( SmartRestDeserializerError::InvalidParameter { ref operation, .. }, - ), + ) + | ref err @ CumulocityMapperError::ExecuteFailed { + operation_name: ref operation, + .. + }, ) => { let topic = C8yTopic::SmartRestResponse.to_topic()?; let msg1 = Message::new(&topic, format!("501,{operation}")); @@ -296,7 +300,6 @@ async fn parse_c8y_topics( error!("{err}"); Ok(vec![msg1, msg2]) } - Err(err) => { error!("{err}"); Ok(vec![]) @@ -576,24 +579,37 @@ async fn validate_and_publish_software_list( Ok(vec![]) } -async fn execute_operation(payload: &str, command: &str) -> Result<(), CumulocityMapperError> { +async fn execute_operation( + payload: &str, + command: &str, + operation_name: &str, +) -> Result<(), CumulocityMapperError> { let command = command.to_owned(); let payload = payload.to_string(); - let _handle = tokio::spawn(async move { - let mut child = tokio::process::Command::new(command) - .args(&[payload]) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .map_err(|e| CumulocityMapperError::ExecuteFailed(e.to_string())) - .unwrap(); - - child.wait().await - }); + let child = tokio::process::Command::new(&command) + .args(&[&payload]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .map_err(|e| CumulocityMapperError::ExecuteFailed { + error_message: e.to_string(), + command: command.to_string(), + operation_name: operation_name.to_string(), + }); - Ok(()) + match child { + Ok(mut child) => { + let _handle = tokio::spawn(async move { + let _ = child.wait().await; + }); + return Ok(()); + } + Err(err) => { + return Err(err); + } + }; } async fn process_smartrest( @@ -660,7 +676,7 @@ async fn forward_operation_request( match operations.matching_smartrest_template(template) { Some(operation) => { if let Some(command) = operation.command() { - execute_operation(payload, command.as_str()).await?; + execute_operation(payload, command.as_str(), &operation.name).await?; } Ok(vec![]) } @@ -717,3 +733,20 @@ pub fn get_child_id_from_topic(topic: &str) -> Result<Option<String>, Conversion option => Ok(option), } } + +mod tests { + #[tokio::test] + async fn test_execute_operation_is_not_blocked() { + let now = std::time::Instant::now(); + let () = super::execute_operation("5", "sleep", "sleep_one") + .await + .unwrap(); + let () = super::execute_operation("5", "sleep", "sleep_two") + .await + .unwrap(); + + // a result between now and elapsed that is not 0 probably means that the operations are + // blocking and that you probably removed a tokio::spawn handle (; + assert_eq!(now.elapsed().as_secs(), 0); + } +} diff --git a/crates/core/tedge_mapper/src/c8y/error.rs b/crates/core/tedge_mapper/src/c8y/error.rs index f9623007..add72d82 100644 --- a/crates/core/tedge_mapper/src/c8y/error.rs +++ b/crates/core/tedge_mapper/src/c8y/error.rs @@ -41,8 +41,12 @@ pub enum CumulocityMapperError { #[error(transparent)] FromIo(#[from] std::io::Error), - #[error("Operation execution failed: {0}")] - ExecuteFailed(String), + #[error("Operation execution failed: {error_message}. Command: {command}. Operation name: {operation_name}")] + ExecuteFailed { + error_message: String, + command: String, + operation_name: String, + }, #[error("An unknown operation template: {0}")] UnknownOperation(String), |