-
Notifications
You must be signed in to change notification settings - Fork 56
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
C8y inventory updates via twin/ topic channel
- Loading branch information
1 parent
12819ca
commit 5a14baa
Showing
8 changed files
with
213 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| use crate::converter::INVENTORY_MANAGED_OBJECTS_TOPIC; | ||
| use serde_json::json; | ||
| use serde_json::Value as JsonValue; | ||
| use tedge_api::mqtt_topics::EntityTopicId; | ||
| use tedge_mqtt_ext::Message; | ||
| use tedge_mqtt_ext::Topic; | ||
| use tracing::warn; | ||
|
|
||
| use crate::{converter::CumulocityConverter, error::ConversionError}; | ||
|
|
||
| impl CumulocityConverter { | ||
| pub fn try_convert_entity_twin_data( | ||
| &mut self, | ||
| source: &EntityTopicId, | ||
| message: &Message, | ||
| fragment_key: &str, | ||
| ) -> Result<Vec<Message>, ConversionError> { | ||
| let target_entity = self.entity_store.try_get(source)?; | ||
| let entity_external_id = target_entity.external_id.as_ref(); | ||
| let payload = serde_json::from_slice::<JsonValue>(message.payload_bytes())?; | ||
|
|
||
| let sanitized_payload = if let JsonValue::Object(mut properties) = payload { | ||
| if properties.contains_key("name") { | ||
| warn!( | ||
| "Updating the entity `name` field via the twin/ topic channel is not supported" | ||
| ); | ||
| properties.remove("name"); | ||
| } | ||
| if properties.contains_key("type") { | ||
| warn!( | ||
| "Updating the entity `type` field via the twin/ topic channel is not supported" | ||
| ); | ||
| properties.remove("name"); | ||
| } | ||
| JsonValue::Object(properties) | ||
| } else { | ||
| payload | ||
| }; | ||
|
|
||
| let mapped_json = json!({ fragment_key: sanitized_payload }); | ||
|
|
||
| let topic = Topic::new_unchecked(&format!( | ||
| "{INVENTORY_MANAGED_OBJECTS_TOPIC}/{entity_external_id}" | ||
| )); | ||
| Ok(vec![Message::new(&topic, mapped_json.to_string())]) | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use crate::converter::tests::create_c8y_converter; | ||
| use serde_json::json; | ||
| use tedge_mqtt_ext::test_helpers::assert_messages_includes_json; | ||
| use tedge_mqtt_ext::{Message, Topic}; | ||
| use tedge_test_utils::fs::TempTedgeDir; | ||
|
|
||
| #[tokio::test] | ||
| async fn convert_entity_twin_data_json_object() { | ||
| let tmp_dir = TempTedgeDir::new(); | ||
| let (mut converter, _http_proxy) = create_c8y_converter(&tmp_dir).await; | ||
|
|
||
| let twin_topic = "te/device/main///twin/device_os"; | ||
| let twin_payload = json!({ | ||
| "family": "Debian", | ||
| "version": "11" | ||
| }); | ||
| let twin_message = | ||
| Message::new(&Topic::new_unchecked(twin_topic), twin_payload.to_string()); | ||
| let inventory_messages = converter.convert(&twin_message).await; | ||
|
|
||
| assert_messages_includes_json( | ||
| inventory_messages, | ||
| [( | ||
| "c8y/inventory/managedObjects/update/test-device", | ||
| json!({ | ||
| "device_os": { | ||
| "family": "Debian", | ||
| "version": "11" | ||
| } | ||
| }), | ||
| )], | ||
| ); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn convert_entity_twin_data_string_value() { | ||
| let tmp_dir = TempTedgeDir::new(); | ||
| let (mut converter, _http_proxy) = create_c8y_converter(&tmp_dir).await; | ||
|
|
||
| let twin_message = Message::new( | ||
| &Topic::new_unchecked("te/device/main///twin/foo"), | ||
| r#""bar""#, | ||
| ); | ||
| let inventory_messages = converter.convert(&twin_message).await; | ||
|
|
||
| assert_messages_includes_json( | ||
| inventory_messages, | ||
| [( | ||
| "c8y/inventory/managedObjects/update/test-device", | ||
| json!({ | ||
| "foo": "bar" | ||
| }), | ||
| )], | ||
| ); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn convert_entity_twin_data_numeric_value() { | ||
| let tmp_dir = TempTedgeDir::new(); | ||
| let (mut converter, _http_proxy) = create_c8y_converter(&tmp_dir).await; | ||
|
|
||
| let twin_message = Message::new( | ||
| &Topic::new_unchecked("te/device/main///twin/foo"), | ||
| r#"5.6789"#, | ||
| ); | ||
| let inventory_messages = converter.convert(&twin_message).await; | ||
|
|
||
| assert_messages_includes_json( | ||
| inventory_messages, | ||
| [( | ||
| "c8y/inventory/managedObjects/update/test-device", | ||
| json!({ | ||
| "foo": 5.6789 | ||
| }), | ||
| )], | ||
| ); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn convert_entity_twin_data_boolean_value() { | ||
| let tmp_dir = TempTedgeDir::new(); | ||
| let (mut converter, _http_proxy) = create_c8y_converter(&tmp_dir).await; | ||
|
|
||
| let twin_message = Message::new( | ||
| &Topic::new_unchecked("te/device/main///twin/enabled"), | ||
| r#"false"#, | ||
| ); | ||
| let inventory_messages = converter.convert(&twin_message).await; | ||
|
|
||
| assert_messages_includes_json( | ||
| inventory_messages, | ||
| [( | ||
| "c8y/inventory/managedObjects/update/test-device", | ||
| json!({ | ||
| "enabled": false | ||
| }), | ||
| )], | ||
| ); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters