From 7e62504db494d9a2523c3001b2ad6d5fa84ce548 Mon Sep 17 00:00:00 2001 From: PradeepKiruvale Date: Fri, 22 Jul 2022 14:31:08 +0530 Subject: Extend events API to support child devices (#1243) * Events for child devices Extend events APIs to send the event messages from external/child device to device twin in the cloud. Signed-off-by: Pradeep Kumar K J --- crates/core/tedge_mapper/src/c8y/converter.rs | 57 +++++++++++++++++++++------ crates/core/tedge_mapper/src/c8y/tests.rs | 4 +- crates/core/tedge_mapper/src/core/error.rs | 3 ++ 3 files changed, 51 insertions(+), 13 deletions(-) (limited to 'crates/core/tedge_mapper/src') diff --git a/crates/core/tedge_mapper/src/c8y/converter.rs b/crates/core/tedge_mapper/src/c8y/converter.rs index fd9223b5..9912a10b 100644 --- a/crates/core/tedge_mapper/src/c8y/converter.rs +++ b/crates/core/tedge_mapper/src/c8y/converter.rs @@ -92,6 +92,7 @@ where "tedge/alarms/+/+", "c8y-internal/alarms/+/+", "tedge/events/+", + "tedge/events/+/+", ] .try_into() .expect("topics that mapper should subscribe to"); @@ -143,6 +144,7 @@ where "tedge/alarms/+/+", "c8y-internal/alarms/+/+", "tedge/events/+", + "tedge/events/+/+", ] .try_into() .expect("topics that mapper should subscribe to"); @@ -185,7 +187,7 @@ where ) -> Result, ConversionError> { let mut vec: Vec = Vec::new(); - let maybe_child_id = get_child_id_from_topic(&input.topic.name)?; + let maybe_child_id = get_child_id_from_measurement_topic(&input.topic.name)?; let c8y_json_payload = match maybe_child_id { Some(child_id) => { // Need to check if the input Thin Edge JSON is valid before adding a child ID to list @@ -224,31 +226,41 @@ where &mut self, input: &Message, ) -> Result, ConversionError> { - let tedge_event = ThinEdgeEvent::try_from(input.topic.name.as_str(), input.payload_str()?)?; + let mut messages = Vec::new(); + + let tedge_event = ThinEdgeEvent::try_from(&input.topic.name, input.payload_str()?)?; + let child_id = tedge_event.source.clone(); + + let need_registration = self.register_external_device(&tedge_event, &mut messages); + let c8y_event = C8yCreateEvent::try_from(tedge_event)?; // If the message doesn't contain any fields other than `text` and `time`, convert to SmartREST let message = if c8y_event.extras.is_empty() { let smartrest_event = Self::serialize_to_smartrest(&c8y_event)?; let smartrest_topic = Topic::new_unchecked(SMARTREST_PUBLISH_TOPIC); - Message::new(&smartrest_topic, smartrest_event) } else { // If the message contains extra fields other than `text` and `time`, convert to Cumulocity JSON let cumulocity_event_json = serde_json::to_string(&c8y_event)?; let json_mqtt_topic = Topic::new_unchecked(C8Y_JSON_MQTT_EVENTS_TOPIC); - Message::new(&json_mqtt_topic, cumulocity_event_json) }; - // If the MQTT message size is well within the Cumulocity MQTT size limit, use MQTT to send the mapped event as well - if input.payload_bytes().len() < self.size_threshold.0 { - Ok(vec![message]) - } else { - // If the message size is larger than the MQTT size limit, use HTTP to send the mapped event + if self.can_send_over_mqtt(&message) { + // The message can be sent via MQTT + messages.push(message); + } else if !need_registration { + // The message must be sent over HTTP let _ = self.http_proxy.send_event(c8y_event).await?; - Ok(vec![]) + return Ok(vec![]); + } else { + // The message should be sent over HTTP but this cannot be done + return Err(ConversionError::ChildDeviceNotRegistered { + id: child_id.unwrap_or_else(|| "".into()), + }); } + Ok(messages) } fn serialize_to_smartrest(c8y_event: &C8yCreateEvent) -> Result { @@ -260,6 +272,29 @@ where c8y_event.time.format(&Rfc3339)? )) } + + fn register_external_device( + &mut self, + tedge_event: &ThinEdgeEvent, + messages: &mut Vec, + ) -> bool { + if let Some(c_id) = tedge_event.source.clone() { + // Create the external source if it does not exists + if !self.children.contains(&c_id) { + self.children.insert(c_id.clone()); + messages.push(Message::new( + &Topic::new_unchecked(SMARTREST_PUBLISH_TOPIC), + format!("101,{c_id},{c_id},thin-edge.io-child"), + )); + return true; + } + } + false + } + + fn can_send_over_mqtt(&self, message: &Message) -> bool { + message.payload_bytes().len() < self.size_threshold.0 + } } #[async_trait] @@ -825,7 +860,7 @@ fn get_inventory_fragments(file_path: &str) -> Result Result, ConversionError> { +pub fn get_child_id_from_measurement_topic(topic: &str) -> Result, ConversionError> { match topic.strip_prefix("tedge/measurements/").map(String::from) { Some(maybe_id) if maybe_id.is_empty() => { Err(ConversionError::InvalidChildId { id: maybe_id }) diff --git a/crates/core/tedge_mapper/src/c8y/tests.rs b/crates/core/tedge_mapper/src/c8y/tests.rs index 78eee84c..03125102 100644 --- a/crates/core/tedge_mapper/src/c8y/tests.rs +++ b/crates/core/tedge_mapper/src/c8y/tests.rs @@ -23,7 +23,7 @@ use tedge_test_utils::fs::TempTedgeDir; use test_case::test_case; use tokio::task::JoinHandle; -use super::converter::{get_child_id_from_topic, CumulocityConverter}; +use super::converter::{get_child_id_from_measurement_topic, CumulocityConverter}; const TEST_TIMEOUT_MS: Duration = Duration::from_millis(5000); const MQTT_HOST: &str = "127.0.0.1"; @@ -640,7 +640,7 @@ async fn convert_two_thin_edge_json_messages_given_different_child_id() { #[test_case("tedge/measurements", None; "invalid child id (parent topic)")] #[test_case("foo/bar", None; "invalid child id (invalid topic)")] fn extract_child_id(in_topic: &str, expected_child_id: Option) { - match get_child_id_from_topic(in_topic) { + match get_child_id_from_measurement_topic(in_topic) { Ok(maybe_id) => assert_eq!(maybe_id, expected_child_id), Err(crate::core::error::ConversionError::InvalidChildId { id }) => { assert_eq!(id, "".to_string()) diff --git a/crates/core/tedge_mapper/src/core/error.rs b/crates/core/tedge_mapper/src/core/error.rs index b703161a..2d2e0163 100644 --- a/crates/core/tedge_mapper/src/core/error.rs +++ b/crates/core/tedge_mapper/src/core/error.rs @@ -104,4 +104,7 @@ pub enum ConversionError { #[error(transparent)] FromOperationLogsError(#[from] plugin_sm::operation_logs::OperationLogsError), + + #[error("The given Child ID '{id}' is not registered with Cumulocity. To send the events to the child device, it has to be registered first.")] + ChildDeviceNotRegistered { id: String }, } -- cgit v1.2.3