diff options
Diffstat (limited to 'crates/core')
-rw-r--r-- | crates/core/c8y_translator/src/serializer.rs | 4 | ||||
-rw-r--r-- | crates/core/tedge/Cargo.toml | 2 | ||||
-rw-r--r-- | crates/core/tedge/src/cli/connect/command.rs | 4 | ||||
-rw-r--r-- | crates/core/tedge_mapper/Cargo.toml | 2 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/c8y/converter.rs | 27 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/c8y/tests.rs | 52 | ||||
-rw-r--r-- | crates/core/tedge_watchdog/src/dummy_watchdog.rs | 3 | ||||
-rw-r--r-- | crates/core/tedge_watchdog/src/error.rs | 4 | ||||
-rw-r--r-- | crates/core/tedge_watchdog/src/main.rs | 13 |
9 files changed, 67 insertions, 44 deletions
diff --git a/crates/core/c8y_translator/src/serializer.rs b/crates/core/c8y_translator/src/serializer.rs index e268235a..fb2d788a 100644 --- a/crates/core/c8y_translator/src/serializer.rs +++ b/crates/core/c8y_translator/src/serializer.rs @@ -48,12 +48,12 @@ impl C8yJsonSerializer { // object referenced by "externalId", instead of root device object // referenced by MQTT client's Device ID. let _ = json.write_key("externalSource"); - let _ = json.write_open_obj(); + json.write_open_obj(); let _ = json.write_key("externalId"); let _ = json.write_str(child_id); let _ = json.write_key("type"); let _ = json.write_str("c8y_Serial"); - let _ = json.write_close_obj(); + json.write_close_obj(); } Self { diff --git a/crates/core/tedge/Cargo.toml b/crates/core/tedge/Cargo.toml index c2589b8a..5f09f9b3 100644 --- a/crates/core/tedge/Cargo.toml +++ b/crates/core/tedge/Cargo.toml @@ -7,6 +7,8 @@ rust-version = "1.58.1" license = "Apache-2.0" readme = "README.md" description = "tedge is the cli tool for thin-edge.io" +homepage = "https://thin-edge.io" +repository = "https://github.com/thin-edge/thin-edge.io" [package.metadata.deb] recommends = "mosquitto" diff --git a/crates/core/tedge/src/cli/connect/command.rs b/crates/core/tedge/src/cli/connect/command.rs index faed5fde..8714bdd7 100644 --- a/crates/core/tedge/src/cli/connect/command.rs +++ b/crates/core/tedge/src/cli/connect/command.rs @@ -489,7 +489,7 @@ fn clean_up( bridge_config: &BridgeConfig, ) -> Result<(), ConnectError> { let path = get_bridge_config_file_path(config_location, bridge_config); - let _ = std::fs::remove_file(&path).or_else(ok_if_not_found)?; + std::fs::remove_file(&path).or_else(ok_if_not_found)?; Ok(()) } @@ -516,7 +516,7 @@ fn write_bridge_config_to_file( .join(TEDGE_BRIDGE_CONF_DIR_PATH); // This will forcefully create directory structure if it doesn't exist, we should find better way to do it, maybe config should deal with it? - let _ = create_directories(&dir_path)?; + create_directories(&dir_path)?; let common_config_path = get_common_mosquitto_config_file_path(config_location, common_mosquitto_config); diff --git a/crates/core/tedge_mapper/Cargo.toml b/crates/core/tedge_mapper/Cargo.toml index 009d3cc6..2032a2e6 100644 --- a/crates/core/tedge_mapper/Cargo.toml +++ b/crates/core/tedge_mapper/Cargo.toml @@ -6,6 +6,8 @@ edition = "2021" rust-version = "1.58.1" license = "Apache-2.0" description = "tedge_mapper is the mapper that translates thin-edge.io data model to c8y/az data model." +homepage = "https://thin-edge.io" +repository = "https://github.com/thin-edge/thin-edge.io" [package.metadata.deb] pre-depends = "tedge" diff --git a/crates/core/tedge_mapper/src/c8y/converter.rs b/crates/core/tedge_mapper/src/c8y/converter.rs index 12f3ec3c..03123697 100644 --- a/crates/core/tedge_mapper/src/c8y/converter.rs +++ b/crates/core/tedge_mapper/src/c8y/converter.rs @@ -274,20 +274,23 @@ where ) -> Result<Vec<Message>, ConversionError> { if topic.name.starts_with("tedge/alarms") { let mut mqtt_messages: Vec<Message> = Vec::new(); - let topic_split: Vec<&str> = topic.name.split('/').collect(); - if topic_split.len() == 5 { - let child_id = topic_split[4]; - // Create a child device, if it does not exists already - if !child_id.is_empty() && !self.children.contains(child_id) { - self.children.insert(child_id.to_string()); - mqtt_messages.push(Message::new( - &Topic::new_unchecked(SMARTREST_PUBLISH_TOPIC), - format!("101,{child_id},{child_id},thin-edge.io-child"), - )); - } - } self.size_threshold.validate(message)?; let mut messages = self.alarm_converter.try_convert_alarm(message)?; + if !messages.is_empty() { + // When there is some messages to be sent on behalf of a child device, + // this child device must be declared first, if not done yet + let topic_split: Vec<&str> = topic.name.split('/').collect(); + if topic_split.len() == 5 { + let child_id = topic_split[4]; + if !child_id.is_empty() && !self.children.contains(child_id) { + self.children.insert(child_id.to_string()); + mqtt_messages.push(Message::new( + &Topic::new_unchecked(SMARTREST_PUBLISH_TOPIC), + format!("101,{child_id},{child_id},thin-edge.io-child"), + )); + } + } + } mqtt_messages.append(&mut messages); Ok(mqtt_messages) } else if topic.name.starts_with(INTERNAL_ALARMS_TOPIC) { diff --git a/crates/core/tedge_mapper/src/c8y/tests.rs b/crates/core/tedge_mapper/src/c8y/tests.rs index 0413c33a..e72f9eed 100644 --- a/crates/core/tedge_mapper/src/c8y/tests.rs +++ b/crates/core/tedge_mapper/src/c8y/tests.rs @@ -76,8 +76,8 @@ async fn mapper_publishes_software_update_request() { let (_tmp_dir, sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); // Prepare and publish a software update smartrest request on `c8y/s/ds`. let smartrest = r#"528,external_id,nodered,1.0.0::debian,,install"#; - let _ = broker.publish("c8y/s/ds", smartrest).await.unwrap(); - let _ = publish_a_fake_jwt_token(broker).await; + broker.publish("c8y/s/ds", smartrest).await.unwrap(); + publish_a_fake_jwt_token(broker).await; let expected_update_list = r#" "updateList": [ @@ -114,7 +114,7 @@ async fn mapper_publishes_software_update_status_onto_c8y_topic() { // Start SM Mapper let (_tmp_dir, sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); - let _ = publish_a_fake_jwt_token(broker).await; + publish_a_fake_jwt_token(broker).await; // Prepare and publish a software update status response message `executing` on `tedge/commands/res/software/update`. let json_response = r#"{ @@ -122,7 +122,7 @@ async fn mapper_publishes_software_update_status_onto_c8y_topic() { "status": "executing" }"#; - let _ = broker + broker .publish("tedge/commands/res/software/update", json_response) .await .unwrap(); @@ -145,7 +145,7 @@ async fn mapper_publishes_software_update_status_onto_c8y_topic() { ]} ]}"#; - let _ = broker + broker .publish("tedge/commands/res/software/update", json_response) .await .unwrap(); @@ -169,7 +169,7 @@ async fn mapper_publishes_software_update_failed_status_onto_c8y_topic() { // Start SM Mapper let (_tmp_dir, sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); - let _ = publish_a_fake_jwt_token(broker).await; + publish_a_fake_jwt_token(broker).await; // The agent publish an error let json_response = r#" @@ -191,7 +191,7 @@ async fn mapper_publishes_software_update_failed_status_onto_c8y_topic() { "failures":[] }"#; - let _ = broker + broker .publish("tedge/commands/res/software/update", json_response) .await .unwrap(); @@ -235,8 +235,8 @@ async fn mapper_fails_during_sw_update_recovers_and_process_response() -> Result // Prepare and publish a software update smartrest request on `c8y/s/ds`. let smartrest = r#"528,external_id,nodered,1.0.0::debian,,install"#; - let _ = broker.publish("c8y/s/ds", smartrest).await.unwrap(); - let _ = publish_a_fake_jwt_token(broker).await; + broker.publish("c8y/s/ds", smartrest).await.unwrap(); + publish_a_fake_jwt_token(broker).await; let expected_update_list = r#" "updateList": [ @@ -278,7 +278,7 @@ async fn mapper_fails_during_sw_update_recovers_and_process_response() -> Result ] } ]}"#; - let _ = broker + broker .publish( "tedge/commands/res/software/update", &remove_whitespace(json_response), @@ -319,7 +319,7 @@ async fn mapper_publishes_software_update_request_with_wrong_action() { let (_tmp_dir, _sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); // Prepare and publish a c8y_SoftwareUpdate smartrest request on `c8y/s/ds` that contains a wrong action `remove`, that is not known by c8y. let smartrest = r#"528,external_id,nodered,1.0.0::debian,,remove"#; - let _ = broker.publish("c8y/s/ds", smartrest).await.unwrap(); + broker.publish("c8y/s/ds", smartrest).await.unwrap(); // Expect a 501 (executing) followed by a 502 (failed) mqtt_tests::assert_received_all_expected( @@ -341,7 +341,7 @@ async fn c8y_mapper_alarm_mapping_to_smartrest() { // Start the C8Y Mapper let (_tmp_dir, sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); - let _ = broker + broker .publish_with_opts( "tedge/alarms/major/temperature_alarm", r#"{ "text": "Temperature high" }"#, @@ -360,7 +360,7 @@ async fn c8y_mapper_alarm_mapping_to_smartrest() { .await; //Clear the previously published alarm - let _ = broker + broker .publish_with_opts( "tedge/alarms/major/temperature_alarm", "", @@ -385,7 +385,7 @@ async fn c8y_mapper_child_alarm_mapping_to_smartrest() { // Start the C8Y Mapper let (_tmp_dir, sm_mapper) = start_c8y_mapper(broker.port).await.unwrap(); - let _ = broker + broker .publish_with_opts( "tedge/alarms/minor/temperature_high/external_sensor", r#"{ "text": "Temperature high" }"#, @@ -395,7 +395,7 @@ async fn c8y_mapper_child_alarm_mapping_to_smartrest() { .await .unwrap(); - let _ = broker + broker .publish_with_opts( "tedge/alarms/minor/temperature_high/external_sensor", r#"{ "text": "Temperature high" }"#, @@ -414,7 +414,7 @@ async fn c8y_mapper_child_alarm_mapping_to_smartrest() { .await; //Clear the previously published alarm - let _ = broker + broker .publish_with_opts( "tedge/alarms/minor/temperature_high/external_sensor", "", @@ -441,7 +441,7 @@ async fn c8y_mapper_syncs_pending_alarms_on_startup() { .messages_published_on("c8y-internal/alarms/critical/temperature_alarm") .await; - let _ = broker + broker .publish_with_opts( "tedge/alarms/critical/temperature_alarm", r#"{ "text": "Temperature very high" }"#, @@ -471,7 +471,7 @@ async fn c8y_mapper_syncs_pending_alarms_on_startup() { sm_mapper.abort(); //Publish a new alarm while the mapper is down - let _ = broker + broker .publish_with_opts( "tedge/alarms/critical/pressure_alarm", r#"{ "text": "Pressure very high" }"#, @@ -483,8 +483,7 @@ async fn c8y_mapper_syncs_pending_alarms_on_startup() { // Ignored until the rumqttd broker bug that doesn't handle empty retained messages //Clear the existing alarm while the mapper is down - // let _ = broker - // .publish_with_opts( + // broker.publish_with_opts( // "tedge/alarms/critical/temperature_alarm", // "", // mqtt_channel::QoS::AtLeastOnce, @@ -533,7 +532,7 @@ async fn c8y_mapper_syncs_pending_child_alarms_on_startup() { .messages_published_on("c8y-internal/alarms/critical/temperature_alarm/external_sensor") .await; - let _ = broker + broker .publish_with_opts( "tedge/alarms/critical/temperature_alarm/external_sensor", r#"{ "text": "Temperature very high" }"#, @@ -563,7 +562,7 @@ async fn c8y_mapper_syncs_pending_child_alarms_on_startup() { sm_mapper.abort(); //Publish a new alarm while the mapper is down - let _ = broker + broker .publish_with_opts( "tedge/alarms/critical/pressure_alarm/external_sensor", r#"{ "text": "Pressure very high" }"#, @@ -573,7 +572,7 @@ async fn c8y_mapper_syncs_pending_child_alarms_on_startup() { .await .unwrap(); - let _ = broker + broker .publish_with_opts( "tedge/alarms/critical/pressure_alarm/external_sensor", r#"{ "text": "Pressure very high" }"#, @@ -585,8 +584,7 @@ async fn c8y_mapper_syncs_pending_child_alarms_on_startup() { // Ignored until the rumqttd broker bug that doesn't handle empty retained messages //Clear the existing alarm while the mapper is down - // let _ = broker - // .publish_with_opts( + // broker.publish_with_opts( // "tedge/alarms/critical/temperature_alarm/external_sensor", // "", // mqtt_channel::QoS::AtLeastOnce, @@ -683,7 +681,7 @@ async fn test_sync_child_alarms() { let alarm_message = Message::new(&Topic::new_unchecked(alarm_topic), alarm_payload); // During the sync phase, alarms are not converted immediately, but only cached to be synced later - assert_eq!(converter.convert(&alarm_message).await.len(), 1); + assert!(converter.convert(&alarm_message).await.is_empty()); let non_alarm_topic = "tedge/measurements/external_sensor"; let non_alarm_payload = r#"{"temp": 1}"#; @@ -1169,5 +1167,5 @@ fn remove_whitespace(s: &str) -> String { } async fn publish_a_fake_jwt_token(broker: &MqttProcessHandler) { - let _ = broker.publish("c8y/s/dat", "71,1111").await.unwrap(); + broker.publish("c8y/s/dat", "71,1111").await.unwrap(); } diff --git a/crates/core/tedge_watchdog/src/dummy_watchdog.rs b/crates/core/tedge_watchdog/src/dummy_watchdog.rs new file mode 100644 index 00000000..f701ffbe --- /dev/null +++ b/crates/core/tedge_watchdog/src/dummy_watchdog.rs @@ -0,0 +1,3 @@ +pub async fn start_watchdog(_config_dir: PathBuf) -> Result<(), anyhow::Error> { + anyhow::Error::from(crate::error::WatchdogError::WatchdogNotAvailable) +} diff --git a/crates/core/tedge_watchdog/src/error.rs b/crates/core/tedge_watchdog/src/error.rs index f9763798..27a9a1ed 100644 --- a/crates/core/tedge_watchdog/src/error.rs +++ b/crates/core/tedge_watchdog/src/error.rs @@ -4,6 +4,10 @@ use tedge_config::{ConfigSettingError, TEdgeConfigError}; #[derive(Debug, thiserror::Error)] pub enum WatchdogError { + #[cfg(not(target_os = "linux"))] + #[error("The watchdog is not available on this platform")] + WatchdogNotAvailable, + #[error("Fail to run `{cmd}`: {from}")] CommandExecError { cmd: String, from: std::io::Error }, diff --git a/crates/core/tedge_watchdog/src/main.rs b/crates/core/tedge_watchdog/src/main.rs index e4478b2f..8f0c2b4b 100644 --- a/crates/core/tedge_watchdog/src/main.rs +++ b/crates/core/tedge_watchdog/src/main.rs @@ -3,7 +3,18 @@ use std::path::PathBuf; use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; mod error; + +// on linux, we use systemd +#[cfg(target_os = "linux")] mod systemd_watchdog; +#[cfg(target_os = "linux")] +use systemd_watchdog as watchdog; + +// on non-linux, we do nothing for now +#[cfg(not(target_os = "linux"))] +mod dummy_watchdog; +#[cfg(not(target_os = "linux"))] +use dummy_watchdog as watchdog; #[derive(Debug, clap::Parser)] #[clap( @@ -31,5 +42,5 @@ async fn main() -> Result<(), anyhow::Error> { let watchdog_opt = WatchdogOpt::parse(); tedge_utils::logging::initialise_tracing_subscriber(watchdog_opt.debug); - systemd_watchdog::start_watchdog(watchdog_opt.config_dir).await + watchdog::start_watchdog(watchdog_opt.config_dir).await } |