feat: c8y-mapper sends all supported operations on-demand via signal channel

Signed-off-by: Rina Fujino <[email protected]>

Author: rina23q

crates/core/tedge_api/src/entity.rs

@@ -13,7 +13,7 @@ use thiserror::Error;
 /// Although this struct doesn't enforce any restrictions for the values,
 /// the consumers may impose restrictions on the accepted values.
 
-#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 #[serde(transparent)]
 pub struct EntityExternalId(String);
 

crates/core/tedge_api/src/mqtt_topics.rs

@@ -179,6 +179,8 @@ impl MqttSchema {
             ChannelFilter::Command(operation) => format!("/cmd/{operation}/+"),
             ChannelFilter::AnyCommandMetadata => "/cmd/+".to_string(),
             ChannelFilter::CommandMetadata(operation) => format!("/cmd/{operation}"),
+            ChannelFilter::AnySignal => "/signal/+".to_string(),
+            ChannelFilter::Signal(signal_type) => format!("/signal/{signal_type}"),
             ChannelFilter::Health => "/status/health".to_string(),
         };
 
@@ -608,6 +610,9 @@ pub enum Channel {
         operation: OperationType,
         cmd_id: String,
     },
+    Signal {
+        signal_type: SignalType,
+    },
     MeasurementMetadata {
         measurement_type: String,
     },
@@ -660,6 +665,9 @@ impl FromStr for Channel {
                 operation: operation.parse().unwrap(), // Infallible
                 cmd_id: cmd_id.to_string(),
             }),
+            ["signal", signal_type] => Ok(Channel::Signal {
+                signal_type: signal_type.into(),
+            }),
             ["status", "health"] => Ok(Channel::Health),
 
             _ => Err(ChannelError::InvalidCategory(channel.to_string())),
@@ -687,6 +695,7 @@ impl Display for Channel {
             Channel::Command { operation, cmd_id } => write!(f, "cmd/{operation}/{cmd_id}"),
             Channel::CommandMetadata { operation } => write!(f, "cmd/{operation}"),
             Channel::Health => write!(f, "status/health"),
+            Channel::Signal { signal_type } => write!(f, "signal/{signal_type}"),
         }
     }
 }
@@ -786,6 +795,64 @@ impl OperationType {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum SignalType {
+    SyncOperations,
+    Custom(String),
+}
+
+// Using a custom Serialize/Deserialize implementations to read "foo" as Custom("foo")
+impl<'de> Deserialize<'de> for SignalType {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let str = String::deserialize(deserializer)?;
+        Ok(str.as_str().into())
+    }
+}
+
+impl Serialize for SignalType {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.to_string().serialize(serializer)
+    }
+}
+
+impl FromStr for SignalType {
+    type Err = Infallible;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(s.into())
+    }
+}
+
+impl<'a> From<&'a str> for SignalType {
+    fn from(s: &'a str) -> SignalType {
+        match s {
+            "sync_operations" => SignalType::SyncOperations,
+            custom => SignalType::Custom(custom.to_string()),
+        }
+    }
+}
+
+impl Display for SignalType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            SignalType::SyncOperations => write!(f, "sync_operations"),
+            SignalType::Custom(custom) => write!(f, "{custom}"),
+        }
+    }
+}
+
+impl SignalType {
+    pub fn name(&self) -> String {
+        format!("{self}")
+    }
+}
+
 #[derive(Debug, thiserror::Error, PartialEq, Eq, Clone)]
 pub enum ChannelError {
     #[error("Channel needs to have at least 2 segments")]
@@ -809,6 +876,8 @@ pub enum ChannelFilter {
     Alarm,
     AnyCommand,
     Command(OperationType),
+    AnySignal,
+    Signal(SignalType),
     MeasurementMetadata,
     EventMetadata,
     AlarmMetadata,
@@ -831,6 +900,7 @@ impl From<&Channel> for ChannelFilter {
                 operation,
                 cmd_id: _,
             } => ChannelFilter::Command(operation.clone()),
+            Channel::Signal { signal_type } => ChannelFilter::Signal(signal_type.clone()),
             Channel::MeasurementMetadata {
                 measurement_type: _,
             } => ChannelFilter::MeasurementMetadata,
@@ -1090,6 +1160,15 @@ mod tests {
             ),
             mqtt_channel::Topic::new_unchecked("te/device/main///cmd/log_upload")
         );
+        assert_eq!(
+            mqtt_schema.topic_for(
+                &device,
+                &Channel::Signal {
+                    signal_type: SignalType::SyncOperations,
+                }
+            ),
+            mqtt_channel::Topic::new_unchecked("te/device/main///signal/sync_operations")
+        );
         assert_eq!(
             mqtt_schema.topic_for(&device, &Channel::Health),
             mqtt_channel::Topic::new_unchecked("te/device/main///status/health")

crates/extensions/c8y_mapper_ext/src/config.rs

@@ -14,6 +14,8 @@ use std::str::FromStr;
 use std::sync::Arc;
 use tedge_api::mqtt_topics::ChannelFilter::AnyCommand;
 use tedge_api::mqtt_topics::ChannelFilter::AnyCommandMetadata;
+use tedge_api::mqtt_topics::ChannelFilter::AnySignal;
+use tedge_api::mqtt_topics::EntityFilter;
 use tedge_api::mqtt_topics::EntityFilter::AnyEntity;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::IdGenerator;
@@ -213,6 +215,10 @@ impl C8yMapperConfig {
         topics.add_all(mqtt_schema.topics(AnyEntity, AnyCommand));
         topics.add_all(mqtt_schema.topics(AnyEntity, AnyCommandMetadata));
 
+        // Add signal topic
+        let mapper_service_tid = EntityTopicId::default_main_service("tedge-mapper-c8y").unwrap();
+        topics.add_all(mqtt_schema.topics(EntityFilter::Entity(&mapper_service_tid), AnySignal));
+
         // Add user configurable external topic filters
         for topic in c8y_config.topics.0.clone() {
             if topics.add(&topic).is_err() {

crates/extensions/c8y_mapper_ext/src/converter.rs

@@ -189,7 +189,7 @@ pub struct CumulocityConverter {
     pub recently_completed_commands: HashMap<CmdId, Instant>,
     active_commands_last_cleared: Instant,
 
-    supported_operations: SupportedOperations,
+    pub supported_operations: SupportedOperations,
     pub operation_handler: OperationHandler,
 
     units: Units,
@@ -1353,6 +1353,8 @@ impl CumulocityConverter {
                 Ok(vec![])
             }
 
+            Channel::Signal { signal_type } => self.process_signal_message(&source, signal_type),
+
             Channel::Health => self.process_health_status_message(&source, message).await,
 
             _ => Ok(vec![]),
@@ -1427,13 +1429,9 @@ impl CumulocityConverter {
     fn try_init_messages(&mut self) -> Result<Vec<MqttMessage>, ConversionError> {
         let mut messages = vec![self.c8y_agent_inventory_fragment()?];
 
-        self.supported_operations
-            .load_all(&self.config.device_id, &self.config.bridge_config)?;
-        let supported_operations_message = self.supported_operations.create_supported_operations(
-            &self.config.device_id,
-            &self.config.bridge_config.c8y_prefix,
-        )?;
-
+        // supported operations for the main device
+        let supported_operations_message =
+            self.load_and_create_supported_operations_messages(&self.config.device_id.clone())?;
         let pending_operations_message =
             create_get_pending_operations_message(&self.config.bridge_config.c8y_prefix);
 
@@ -1444,6 +1442,19 @@ impl CumulocityConverter {
         Ok(messages)
     }
 
+    pub fn load_and_create_supported_operations_messages(
+        &mut self,
+        external_id: &str,
+    ) -> Result<MqttMessage, ConversionError> {
+        self.supported_operations
+            .load_all(external_id, &self.config.bridge_config)?;
+        let supported_operations_message = self
+            .supported_operations
+            .create_supported_operations(external_id, &self.config.bridge_config.c8y_prefix)?;
+
+        Ok(supported_operations_message)
+    }
+
     pub fn sync_messages(&mut self) -> Vec<MqttMessage> {
         let sync_messages: Vec<MqttMessage> = self.alarm_converter.sync();
         self.alarm_converter = AlarmConverter::Synced;

crates/extensions/c8y_mapper_ext/src/entity_cache.rs

@@ -386,6 +386,13 @@ impl EntityCache {
     pub fn cache_early_data_message(&mut self, message: MqttMessage) {
         self.pending_entities.cache_early_data_message(message)
     }
+
+    pub fn get_all_external_ids(&self) -> Vec<EntityExternalId> {
+        let mut ids: Vec<EntityExternalId> =
+            self.external_id_map.keys().map(Clone::clone).collect();
+        ids.sort();
+        ids
+    }
 }
 
 #[cfg(test)]

crates/extensions/c8y_mapper_ext/src/lib.rs

@@ -13,6 +13,7 @@ pub mod json;
 mod operations;
 mod serializer;
 pub mod service_monitor;
+mod signals;
 mod supported_operations;
 
 #[cfg(test)]

crates/extensions/c8y_mapper_ext/src/signals.rs

@@ -0,0 +1,34 @@
+use crate::converter::CumulocityConverter;
+use crate::error::ConversionError;
+use tedge_api::mqtt_topics::EntityTopicId;
+use tedge_api::mqtt_topics::SignalType;
+use tedge_mqtt_ext::MqttMessage;
+
+impl CumulocityConverter {
+    pub fn process_signal_message(
+        &mut self,
+        source: &EntityTopicId,
+        signal_type: &SignalType,
+    ) -> Result<Vec<MqttMessage>, ConversionError> {
+        let mut messages = Vec::new();
+
+        if source.default_service_name() != Some("tedge-mapper-c8y") {
+            return Ok(messages);
+        }
+
+        match signal_type {
+            SignalType::SyncOperations => {
+                for external_id in self.entity_cache.get_all_external_ids() {
+                    if let Ok(message) =
+                        self.load_and_create_supported_operations_messages(external_id.as_ref())
+                    {
+                        messages.push(message);
+                    }
+                }
+            }
+            SignalType::Custom(_) => {}
+        }
+
+        Ok(messages)
+    }
+}

crates/extensions/c8y_mapper_ext/src/supported_operations/mod.rs

@@ -48,6 +48,7 @@ type OperationName = String;
 type OperationNameRef = str;
 
 /// Used to hold and query supported operations for all devices.
+#[derive(Debug)]
 pub struct SupportedOperations {
     /// External ID of the main device.
     ///

crates/extensions/c8y_mapper_ext/src/tests.rs

@@ -33,6 +33,8 @@ use tedge_actors::NoMessage;
 use tedge_actors::Sender;
 use tedge_actors::SimpleMessageBox;
 use tedge_actors::SimpleMessageBoxBuilder;
+use tedge_api::mqtt_topics::ChannelFilter::AnySignal;
+use tedge_api::mqtt_topics::EntityFilter::AnyEntity;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
 use tedge_config::models::AutoLogUpload;
@@ -3113,6 +3115,87 @@ async fn mapper_converts_config_cmd_to_supported_op_and_types_for_child_device()
     .await;
 }
 
+#[tokio::test]
+async fn mapper_publishes_all_supported_operations_on_signal() {
+    let ttd = TempTedgeDir::new();
+    ttd.dir("operations").dir("c8y").file("c8y_Restart");
+    ttd.dir("operations")
+        .dir("c8y")
+        .dir("test-device:device:child01");
+    ttd.dir("operations")
+        .dir("c8y")
+        .dir("test-device:device:child02")
+        .file("c8y_Restart");
+    ttd.dir("operations")
+        .dir("c8y")
+        .dir("test-device:device:child02")
+        .file("c8y_UploadConfigFile");
+    ttd.dir("operations")
+        .dir("c8y")
+        .dir("test-device:device:main:service:collectd")
+        .file("c8y_Restart");
+
+    let test_handle = spawn_c8y_mapper_actor(&ttd, true).await;
+    let TestHandle { mqtt, .. } = test_handle;
+    let mut mqtt = mqtt.with_timeout(TEST_TIMEOUT_MS);
+
+    skip_init_messages(&mut mqtt).await;
+
+    // Register the tedge-mapper-c8y service upfront
+    mqtt.send(MqttMessage::new(
+        &Topic::new_unchecked("te/device/main/service/tedge-mapper-c8y"),
+        r#"{"@parent":"device/main//","@type":"service","name":"tedge-mapper-c8y","type":"service"}"#,
+    ))
+    .await
+    .expect("Send failed");
+    mqtt.skip(1).await; // Skip 102 message
+
+    // Also register other entities
+    mqtt.send(MqttMessage::new(
+        &Topic::new_unchecked("te/device/child01//"),
+        r#"{"@parent":"device/main//","@type":"child-device"}"#,
+    ))
+    .await
+    .expect("Send failed");
+    mqtt.send(MqttMessage::new(
+        &Topic::new_unchecked("te/device/child02//"),
+        r#"{"@parent":"device/main//","@type":"child-device"}"#,
+    ))
+    .await
+    .expect("Send failed");
+    mqtt.send(MqttMessage::new(
+        &Topic::new_unchecked("te/device/main/service/collectd"),
+        r#"{"@parent":"device/main//","@type":"service"}"#,
+    ))
+    .await
+    .expect("Send failed");
+    mqtt.skip(3).await; // Skip registration messages
+
+    mqtt.send(MqttMessage::new(
+        &Topic::new_unchecked("te/device/main/service/tedge-mapper-c8y/signal/sync_operations"),
+        "{}",
+    ))
+    .await
+    .expect("Send failed");
+
+    assert_received_contains_str(
+        &mut mqtt,
+        [
+            ("c8y/s/us", "114,c8y_Restart"),
+            ("c8y/s/us/test-device:device:child01", "114"),
+            (
+                "c8y/s/us/test-device:device:child02",
+                "114,c8y_Restart,c8y_UploadConfigFile",
+            ),
+            (
+                "c8y/s/us/test-device:device:main:service:collectd",
+                "114,c8y_Restart",
+            ),
+        ],
+    )
+    .await;
+}
+
 fn assert_command_exec_log_content(cfg_dir: TempTedgeDir, expected_contents: &str) {
     let paths = fs::read_dir(cfg_dir.to_path_buf().join("agent")).unwrap();
     for path in paths {
@@ -3283,6 +3366,7 @@ pub(crate) fn test_mapper_config(tmp_dir: &TempTedgeDir) -> C8yMapperConfig {
     topics.add_all(operation_topics);
 
     topics.add_all(C8yMapperConfig::default_external_topic_filter());
+    topics.add_all(mqtt_schema.topics(AnyEntity, AnySignal));
 
     topics.remove_overlapping_patterns();