Merge pull request #2716 from jarhodes314/2592-mapper-bridge

Add support for built-in bridge to tedge-mapper-c8y

Author: jarhodes314

Cargo.lock

@@ -144,6 +144,7 @@ tedge_file_system_ext = { path = "crates/extensions/tedge_file_system_ext" }
 tedge_health_ext = { path = "crates/extensions/tedge_health_ext" }
 tedge_http_ext = { path = "crates/extensions/tedge_http_ext" }
 tedge_log_manager = { path = "crates/extensions/tedge_log_manager" }
+tedge_mqtt_bridge = { path = "crates/extensions/tedge_mqtt_bridge" }
 tedge_mqtt_ext = { path = "crates/extensions/tedge_mqtt_ext" }
 tedge_script_ext = { path = "crates/extensions/tedge_script_ext" }
 tedge_signal_ext = { path = "crates/extensions/tedge_signal_ext" }

Cargo.toml

@@ -14,13 +14,20 @@ use camino::Utf8PathBuf;
 use certificate::CertificateError;
 use certificate::PemCertificate;
 use doku::Document;
+use doku::Type;
 use once_cell::sync::Lazy;
+use serde::Deserialize;
 use std::borrow::Cow;
+use std::convert::Infallible;
+use std::fmt;
+use std::fmt::Formatter;
 use std::io::Read;
 use std::net::IpAddr;
 use std::net::Ipv4Addr;
 use std::num::NonZeroU16;
+use std::ops::Deref;
 use std::path::PathBuf;
+use std::str::FromStr;
 use std::sync::Arc;
 use tedge_config_macros::all_or_nothing;
 use tedge_config_macros::define_tedge_config;
@@ -444,7 +451,19 @@ define_tedge_config! {
                 #[tedge_config(note = "If set to 'auto', this cleans the local session accordingly the detected version of mosquitto.")]
                 #[tedge_config(example = "auto", default(variable = "AutoFlag::Auto"))]
                 local_cleansession: AutoFlag,
-            }
+            },
+
+            #[tedge_config(default(value = false))]
+            #[doku(skip)] // Hide the configuration in `tedge config list --doc`
+            built_in: bool,
+
+            // TODO validation
+            /// The topic prefix that will be used for the mapper bridge MQTT topic. For instance,
+            /// if this is set to "c8y", then messages published to `c8y/s/us` will be
+            /// forwarded by to Cumulocity on the `s/us` topic
+            #[tedge_config(example = "c8y", default(value = "c8y"))]
+            #[doku(skip)] // Hide the configuration in `tedge config list --doc`
+            topic_prefix: TopicPrefix,
         },
 
         entity_store: {
@@ -798,6 +817,80 @@ define_tedge_config! {
 
 }
 
+impl ReadableKey {
+    // This is designed to be a simple way of controlling whether values appear in the output of
+    // `tedge config list`. Ideally this would be integrated into [define_tedge_config], see
+    // https://github.com/thin-edge/thin-edge.io/issues/2767 for more detail on that.
+    // Currently this accompanies `#[doku(skip)]` on the relevant configurations, which hides
+    // them in `tedge config list --doc`. The configurations are hidden to avoid unfinished
+    // features from being discovered.
+    pub fn is_printable_value(self, value: &str) -> bool {
+        match self {
+            Self::C8yBridgeBuiltIn => value != "false",
+            Self::C8yBridgeTopicPrefix => value != "c8y",
+            _ => true,
+        }
+    }
+}
+
+// TODO doc comment
+#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Deserialize, serde::Serialize)]
+#[serde(from = "String", into = "Arc<str>")]
+pub struct TopicPrefix(Arc<str>);
+
+impl Document for TopicPrefix {
+    fn ty() -> Type {
+        String::ty()
+    }
+}
+
+// TODO actual validation
+// TODO make sure we don't allow c8y-internal either, or az, or aws as those are all used
+impl From<String> for TopicPrefix {
+    fn from(value: String) -> Self {
+        Self(value.into())
+    }
+}
+
+impl From<&str> for TopicPrefix {
+    fn from(value: &str) -> Self {
+        Self(value.into())
+    }
+}
+
+impl FromStr for TopicPrefix {
+    type Err = Infallible;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(s.into())
+    }
+}
+
+impl From<TopicPrefix> for Arc<str> {
+    fn from(value: TopicPrefix) -> Self {
+        value.0
+    }
+}
+
+// TODO is deref actually right here
+impl Deref for TopicPrefix {
+    type Target = str;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl TopicPrefix {
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for TopicPrefix {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
 fn default_http_bind_address(dto: &TEdgeConfigDto) -> IpAddr {
     let external_address = dto.mqtt.external.bind.address;
     external_address
@@ -817,7 +910,8 @@ fn device_id(reader: &TEdgeConfigReader) -> Result<String, ReadError> {
 fn cert_error_into_config_error(key: &'static str, err: CertificateError) -> ReadError {
     match &err {
         CertificateError::IoError(io_err) => match io_err.kind() {
-            std::io::ErrorKind::NotFound => ReadError::ReadOnlyNotFound { key,
+            std::io::ErrorKind::NotFound => ReadError::ReadOnlyNotFound {
+                key,
                 message: concat!(
                     "The device id is read from the device certificate.\n",
                     "To set 'device.id' to some <id>, you can use `tedge cert create --device-id <id>`.",

crates/common/tedge_config/src/tedge_config_cli/tedge_config.rs

@@ -10,6 +10,7 @@ use std::collections::HashMap;
 use std::time::Duration;
 use tedge_config::mqtt_config::MqttConfigBuildError;
 use tedge_config::TEdgeConfig;
+use tedge_config::TopicPrefix;
 use tracing::error;
 use tracing::info;
 
@@ -112,33 +113,41 @@ impl C8yEndPoint {
 
 pub struct C8yMqttJwtTokenRetriever {
     mqtt_config: mqtt_channel::Config,
+    topic_prefix: TopicPrefix,
 }
 
 impl C8yMqttJwtTokenRetriever {
     pub fn from_tedge_config(tedge_config: &TEdgeConfig) -> Result<Self, MqttConfigBuildError> {
         let mqtt_config = tedge_config.mqtt_config()?;
 
-        Ok(Self::new(mqtt_config))
+        Ok(Self::new(
+            mqtt_config,
+            tedge_config.c8y.bridge.topic_prefix.clone(),
+        ))
     }
 
-    pub fn new(mqtt_config: mqtt_channel::Config) -> Self {
-        let topic = TopicFilter::new_unchecked("c8y/s/dat");
+    pub fn new(mqtt_config: mqtt_channel::Config, topic_prefix: TopicPrefix) -> Self {
+        let topic = TopicFilter::new_unchecked(&format!("{topic_prefix}/s/dat"));
         let mqtt_config = mqtt_config
             .with_no_session() // Ignore any already published tokens, possibly stale.
             .with_subscriptions(topic);
 
-        C8yMqttJwtTokenRetriever { mqtt_config }
+        C8yMqttJwtTokenRetriever {
+            mqtt_config,
+            topic_prefix,
+        }
     }
 
     pub async fn get_jwt_token(&mut self) -> Result<SmartRestJwtResponse, JwtError> {
         let mut mqtt_con = Connection::new(&self.mqtt_config).await?;
+        let pub_topic = format!("{}/s/uat", self.topic_prefix);
 
         tokio::time::sleep(Duration::from_millis(20)).await;
         for _ in 0..3 {
             mqtt_con
                 .published
                 .publish(
-                    mqtt_channel::Message::new(&Topic::new_unchecked("c8y/s/uat"), "".to_string())
+                    mqtt_channel::Message::new(&Topic::new_unchecked(&pub_topic), "".to_string())
                         .with_qos(mqtt_channel::QoS::AtMostOnce),
                 )
                 .await?;
@@ -184,7 +193,6 @@ pub enum JwtError {
 
 #[cfg(test)]
 mod tests {
-
     use super::*;
     use test_case::test_case;
 

crates/core/c8y_api/src/http_proxy.rs

@@ -7,21 +7,22 @@ use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::SoftwareModule;
 use tedge_api::SoftwareModuleUpdate;
 use tedge_api::SoftwareUpdateCommand;
+use tedge_config::TopicPrefix;
 use time::OffsetDateTime;
 
 pub struct C8yDeviceControlTopic;
 
 impl C8yDeviceControlTopic {
-    pub fn topic() -> Topic {
-        Topic::new_unchecked(Self::name())
+    pub fn topic(prefix: &TopicPrefix) -> Topic {
+        Topic::new_unchecked(&Self::name(prefix))
     }
 
-    pub fn accept(topic: &Topic) -> bool {
-        topic.name.starts_with(Self::name())
+    pub fn accept(topic: &Topic, prefix: &TopicPrefix) -> bool {
+        topic.name.starts_with(&Self::name(prefix))
     }
 
-    pub fn name() -> &'static str {
-        "c8y/devicecontrol/notifications"
+    pub fn name(prefix: &TopicPrefix) -> String {
+        format!("{prefix}/devicecontrol/notifications")
     }
 }
 
@@ -422,10 +423,15 @@ pub trait C8yDeviceControlOperationHelper {
 }
 
 impl C8yDeviceControlOperationHelper for C8yRestart {}
+
 impl C8yDeviceControlOperationHelper for C8ySoftwareUpdate {}
+
 impl C8yDeviceControlOperationHelper for C8yLogfileRequest {}
+
 impl C8yDeviceControlOperationHelper for C8yUploadConfigFile {}
+
 impl C8yDeviceControlOperationHelper for C8yDownloadConfigFile {}
+
 impl C8yDeviceControlOperationHelper for C8yFirmware {}
 
 #[derive(thiserror::Error, Debug)]

crates/core/c8y_api/src/json_c8y_deserializer.rs

@@ -12,6 +12,7 @@
 use crate::smartrest::csv::fields_to_csv_string;
 use crate::smartrest::topic::publish_topic_from_ancestors;
 use mqtt_channel::Message;
+use tedge_config::TopicPrefix;
 
 use super::message::sanitize_for_smartrest;
 
@@ -23,6 +24,7 @@ pub fn child_device_creation_message(
     device_name: Option<&str>,
     device_type: Option<&str>,
     ancestors: &[String],
+    prefix: &TopicPrefix,
 ) -> Result<Message, InvalidValueError> {
     if child_id.is_empty() {
         return Err(InvalidValueError {
@@ -44,7 +46,7 @@ pub fn child_device_creation_message(
     }
 
     Ok(Message::new(
-        &publish_topic_from_ancestors(ancestors),
+        &publish_topic_from_ancestors(ancestors, prefix),
         // XXX: if any arguments contain commas, output will be wrong
         format!(
             "101,{},{},{}",
@@ -64,6 +66,7 @@ pub fn service_creation_message(
     service_type: &str,
     service_status: &str,
     ancestors: &[String],
+    prefix: &TopicPrefix,
 ) -> Result<Message, InvalidValueError> {
     // TODO: most of this noise can be eliminated by implementing `Serialize`/`Deserialize` for smartrest format
     if service_id.is_empty() {
@@ -92,7 +95,7 @@ pub fn service_creation_message(
     }
 
     Ok(Message::new(
-        &publish_topic_from_ancestors(ancestors),
+        &publish_topic_from_ancestors(ancestors, prefix),
         fields_to_csv_string(&[
             "102",
             service_id,
@@ -116,8 +119,9 @@ pub fn service_creation_message(
 pub fn service_status_update_message(
     external_ids: &[impl AsRef<str>],
     service_status: &str,
+    prefix: &TopicPrefix,
 ) -> Message {
-    let topic = publish_topic_from_ancestors(external_ids);
+    let topic = publish_topic_from_ancestors(external_ids, prefix);
 
     let service_status =
         sanitize_for_smartrest(service_status, super::message::MAX_PAYLOAD_LIMIT_IN_BYTES);

crates/core/c8y_api/src/smartrest/inventory.rs

@@ -7,6 +7,7 @@ use serde::ser::SerializeSeq;
 use serde::Deserialize;
 use serde::Serialize;
 use serde::Serializer;
+use tedge_config::TopicPrefix;
 use tracing::warn;
 
 pub type SmartRest = String;
@@ -204,20 +205,20 @@ where
 
 /// Helper to generate a SmartREST operation status message
 pub trait OperationStatusMessage {
-    fn executing() -> Message {
-        Self::create_message(Self::status_executing())
+    fn executing(prefix: &TopicPrefix) -> Message {
+        Self::create_message(Self::status_executing(), prefix)
     }
 
-    fn successful(parameter: Option<&str>) -> Message {
-        Self::create_message(Self::status_successful(parameter))
+    fn successful(parameter: Option<&str>, prefix: &TopicPrefix) -> Message {
+        Self::create_message(Self::status_successful(parameter), prefix)
     }
 
-    fn failed(failure_reason: &str) -> Message {
-        Self::create_message(Self::status_failed(failure_reason))
+    fn failed(failure_reason: &str, prefix: &TopicPrefix) -> Message {
+        Self::create_message(Self::status_failed(failure_reason), prefix)
     }
 
-    fn create_message(payload: SmartRest) -> Message {
-        let topic = C8yTopic::SmartRestResponse.to_topic().unwrap(); // never fail
+    fn create_message(payload: SmartRest, prefix: &TopicPrefix) -> Message {
+        let topic = C8yTopic::SmartRestResponse.to_topic(prefix).unwrap(); // never fail
         Message::new(&topic, payload)
     }