Merge pull request #3466 from didier-wenzek/refactor/async-tedge-cli

refactor: make tedge cli async

Author: didier-wenzek

.config/nextest.toml

@@ -2,7 +2,15 @@
 retries = 0
 test-threads = "num-cpus"
 
+[test-groups]
+serial = { max-threads = 1 }
+
 [[profile.default.overrides]]
 # Remove once https://github.com/thin-edge/thin-edge.io/issues/3030 is resolved
 filter = 'test(uploaded_file_can_be_downloaded_from_the_api)'
 retries = 4
+
+[[profile.default.overrides]]
+filter = 'package(c8y_firmware_manager)'
+test-group = 'serial'
+

Cargo.lock

@@ -38,7 +38,6 @@ collectd_ext = { path = "crates/extensions/collectd_ext" }
 download = { path = "crates/common/download" }
 flockfile = { path = "crates/common/flockfile" }
 json-writer = { path = "crates/common/json_writer" }
-log_manager = { path = "crates/common/log_manager" }
 mqtt_channel = { path = "crates/common/mqtt_channel" }
 mqtt_tests = { path = "crates/tests/mqtt_tests" }
 plugin_sm = { path = "crates/core/plugin_sm" }
@@ -77,7 +76,7 @@ assert-json-diff = "2.0"
 assert_cmd = "2.0"
 assert_matches = "1.5"
 async-compat = "0.2.1"
-async-log = "2.0"
+async-tempfile = "0.7"
 async-trait = "0.1"
 async-tungstenite = { version = "0.28", features = [
     "tokio-runtime",
@@ -110,7 +109,6 @@ figment = { version = "0.10" }
 filetime = "0.2"
 freedesktop_entry_parser = "1.3.0"
 futures = "0.3"
-futures-timer = "3.0"
 futures-util = "0.3.25"
 glob = "0.3"
 heck = "0.4.1"
@@ -171,8 +169,6 @@ serde_json = "1.0"
 sha-1 = "0.10"
 sha256 = "1.1"
 shell-words = "1.1"
-signal-hook = "0.3"
-signal-hook-tokio = { version = "0.3", features = ["futures-v0_3"] }
 strum = "0.24"
 strum_macros = "0.24"
 syn = { version = "2", features = ["full", "extra-traits"] }
@@ -189,7 +185,6 @@ tower = "0.4"
 tracing = { version = "0.1", features = ["attributes", "log"] }
 tracing-subscriber = { version = "0.3", features = ["time", "env-filter"] }
 try-traits = "0.1"
-tungstenite = "0.26.1"
 url = "2.3"
 uzers = "0.11"
 walkdir = "2"

Cargo.toml

@@ -10,7 +10,6 @@ repository = { workspace = true }
 
 [features]
 default = []
-reqwest-blocking = ["dep:reqwest", "reqwest/blocking"]
 reqwest = ["dep:reqwest"]
 cryptoki = ["dep:cryptoki"]
 

crates/common/certificate/Cargo.toml

@@ -27,25 +27,6 @@ impl CloudRootCerts {
             .build()
             .expect("Valid reqwest client builder configuration")
     }
-
-    #[allow(clippy::disallowed_types)]
-    #[cfg(feature = "reqwest-blocking")]
-    pub fn blocking_client_builder(&self) -> reqwest::blocking::ClientBuilder {
-        self.certificates
-            .iter()
-            .cloned()
-            .fold(reqwest::blocking::ClientBuilder::new(), |builder, cert| {
-                builder.add_root_certificate(cert)
-            })
-    }
-
-    #[allow(clippy::disallowed_types)]
-    #[cfg(feature = "reqwest-blocking")]
-    pub fn blocking_client(&self) -> reqwest::blocking::Client {
-        self.blocking_client_builder()
-            .build()
-            .expect("Valid reqwest client builder configuration")
-    }
 }
 
 impl From<Arc<[Certificate]>> for CloudRootCerts {

crates/common/certificate/src/cloud_root_certificate.rs

@@ -267,7 +267,7 @@ impl Config {
         &mut self,
         ca_file: impl AsRef<Path>,
     ) -> Result<&mut Self, certificate::CertificateError> {
-        debug!("Using CA certificate: {}", ca_file.as_ref().display());
+        debug!(target: "MQTT", "Using CA certificate: {}", ca_file.as_ref().display());
         let authentication_config = self.broker.authentication.get_or_insert(Default::default());
         let cert_store = &mut authentication_config.cert_store;
 
@@ -282,7 +282,7 @@ impl Config {
         &mut self,
         ca_dir: impl AsRef<Path>,
     ) -> Result<&mut Self, certificate::CertificateError> {
-        debug!("Using CA directory: {}", ca_dir.as_ref().display());
+        debug!(target: "MQTT", "Using CA directory: {}", ca_dir.as_ref().display());
         let authentication_config = self.broker.authentication.get_or_insert(Default::default());
         let cert_store = &mut authentication_config.cert_store;
 
@@ -300,8 +300,8 @@ impl Config {
         cert_file: P,
         key_file: P,
     ) -> Result<&mut Self, CertificateError> {
-        debug!("Using client certificate: {}", cert_file.as_ref().display());
-        debug!("Using client private key: {}", key_file.as_ref().display());
+        debug!(target: "MQTT", "Using client certificate: {}", cert_file.as_ref().display());
+        debug!(target: "MQTT", "Using client private key: {}", key_file.as_ref().display());
         let cert_chain = parse_root_certificate::read_cert_chain(cert_file)?;
         let key = parse_root_certificate::read_pvt_key(key_file)?;
 

crates/common/mqtt_channel/src/config.rs

@@ -10,6 +10,7 @@ use futures::SinkExt;
 use futures::StreamExt;
 use log::error;
 use log::info;
+use log::warn;
 use rumqttc::AsyncClient;
 use rumqttc::Event;
 use rumqttc::EventLoop;
@@ -132,17 +133,17 @@ impl Connection {
         const SECURE_MQTT_PORT: u16 = 8883;
 
         if config.broker.port == INSECURE_MQTT_PORT && config.broker.authentication.is_some() {
-            eprintln!("WARNING: Connecting on port 1883 for insecure MQTT using a TLS connection");
+            warn!(target: "MQTT", "Connecting on port 1883 for insecure MQTT using a TLS connection");
         }
         if config.broker.port == SECURE_MQTT_PORT && config.broker.authentication.is_none() {
-            eprintln!("WARNING: Connecting on port 8883 for secure MQTT without a CA file");
+            warn!(target: "MQTT", "Connecting on port 8883 for secure MQTT without a CA file");
         }
 
         let mqtt_options = config.rumqttc_options()?;
         let (mqtt_client, mut event_loop) = AsyncClient::new(mqtt_options, config.queue_capacity);
 
-        info!(
-            "MQTT connecting to broker: host={}:{}, session_name={:?}",
+        info!(target: "MQTT",
+            "Connecting to broker: host={}:{}, session_name={:?}",
             config.broker.host, config.broker.port, config.session_name
         );
 
@@ -152,7 +153,7 @@ impl Connection {
                     if let Some(err) = MqttError::maybe_connection_error(&ack) {
                         return Err(err);
                     };
-                    info!("MQTT connection established");
+                    info!(target: "MQTT", "Connection established");
 
                     let subscriptions = config.subscriptions.filters();
 
@@ -175,16 +176,16 @@ impl Connection {
                     // Messages can be received before a sub ack
                     // Errors on send are ignored: it just means the client has closed the receiving channel.
                     if msg.payload.len() > config.max_packet_size {
-                        error!("Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}", 
+                        error!(target: "MQTT", "Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}",
                             msg.topic, msg.payload.len(), config.max_packet_size);
                         continue;
                     }
                     let _ = message_sender.send(msg.into()).await;
                 }
 
                 Err(err) => {
-                    error!(
-                        "MQTT: failed to connect to broker at '{host}:{port}': {err}",
+                    error!(target: "MQTT",
+                        "Failed to connect to broker at '{host}:{port}': {err}",
                         host = config.broker.host,
                         port = config.broker.port
                     );
@@ -244,7 +245,7 @@ impl Connection {
             match event {
                 Ok(Event::Incoming(Packet::Publish(msg))) => {
                     if msg.payload.len() > config.max_packet_size {
-                        error!("Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}", 
+                        error!(target: "MQTT", "Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}",
                             msg.topic, msg.payload.len(), config.max_packet_size);
                         continue;
                     }
@@ -255,9 +256,9 @@ impl Connection {
 
                 Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                     if let Some(err) = MqttError::maybe_connection_error(&ack) {
-                        error!("MQTT connection Error {err}");
+                        error!(target: "MQTT", "Connection Error {err}");
                     } else {
-                        info!("MQTT connection re-established");
+                        info!(target: "MQTT", "Connection re-established");
                         if let Some(ref imsg_fn) = config.initial_message {
                             // publish the initial message on connect
                             let message = imsg_fn.new_init_message();
@@ -288,12 +289,11 @@ impl Connection {
 
                 Ok(Event::Incoming(Incoming::Disconnect))
                 | Ok(Event::Outgoing(Outgoing::Disconnect)) => {
-                    info!("MQTT connection closed");
                     break;
                 }
 
                 Err(err) => {
-                    error!("MQTT connection error: {err}");
+                    error!(target: "MQTT", "Connection error: {err}");
 
                     // Errors on send are ignored: it just means the client has closed the receiving channel.
                     let _ = error_sender.send(err.into()).await;
@@ -303,6 +303,16 @@ impl Connection {
                 _ => (),
             }
         }
+
+        // Wait for Err(MqttState(ConnectionAborted))
+        // to make sure the disconnect is effective
+        loop {
+            if (event_loop.poll().await).is_err() {
+                info!(target: "MQTT", "Connection closed");
+                break;
+            }
+        }
+
         // No more messages will be forwarded to the client
         let _ = message_sender.close().await;
         let _ = error_sender.close().await;

crates/common/mqtt_channel/src/connection.rs

@@ -184,6 +184,10 @@ impl MqttMessage {
         }
     }
 
+    pub fn with_retain_flag(self, retain: bool) -> Self {
+        Self { retain, ..self }
+    }
+
     /// The message payload
     pub fn payload(&self) -> &Payload {
         &self.payload.0

crates/common/mqtt_channel/src/messages.rs

@@ -1,5 +1,6 @@
 use crate::Config;
 use crate::MqttError;
+use log::error;
 use log::warn;
 use rumqttc::AsyncClient;
 use rumqttc::ConnectReturnCode;
@@ -44,7 +45,7 @@ pub async fn init_session(config: &Config) -> Result<(), MqttError> {
             Err(err) => match err {
                 rumqttc::ConnectionError::ConnectionRefused(ConnectReturnCode::Success) => {}
                 _ => {
-                    warn!("{}", MqttError::from_connection_error(err));
+                    warn!(target: "MQTT", "{}", MqttError::from_connection_error(err));
                     break;
                 }
             },
@@ -85,7 +86,7 @@ pub async fn clear_session(config: &Config) -> Result<(), MqttError> {
             }
 
             Err(err) => {
-                eprintln!("Connection Error {}", err);
+                error!(target: "MQTT", "Connection Error {}", err);
                 break;
             }
             _ => (),

crates/common/mqtt_channel/src/session.rs

@@ -34,6 +34,7 @@ strum_macros = { workspace = true }
 tedge_config_macros = { workspace = true }
 tedge_utils = { workspace = true, features = ["timestamp"] }
 thiserror = { workspace = true }
+tokio = { workspace = true }
 toml = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }