Connect mqtt5 transport at startup

Author: AnotherDaniel

Cargo.lock

@@ -30,11 +30,17 @@ version = "0.4.0"
 
 [workspace.dependencies]
 async-trait = { version = "0.1" }
+backon = { version = "1.6", default-features = false, features = [
+    "tokio-sleep",
+] }
 mockall = { version = "0.14" }
 protobuf = { version = "3.7.2" }
 test-case = { version = "3.3" }
 tokio = { version = "1", features = ["full"] }
-tracing = { version = "0.1", default-features = false, features = ["log", "std"] }
+tracing = { version = "0.1", default-features = false, features = [
+    "log",
+    "std",
+] }
 up-rust = { version = "0.9.0", features = ["usubscription"] }
 up-subscription = { path = "./up-subscription" }
 

Cargo.toml

@@ -33,6 +33,7 @@ mqtt5 = ["dep:up-transport-mqtt5"]
 zenoh = ["dep:up-transport-zenoh", "dep:serde_json"]
 
 [dependencies]
+backon = { workspace = true }
 clap = { version = "4.5.53", default-features = false, features = [
     "std",
     "derive",
@@ -47,7 +48,7 @@ serde_json = { version = "1.0", optional = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
-up-rust = { workspace = true, features = ["usubscription", "util" ] }
+up-rust = { workspace = true, features = ["usubscription", "util"] }
 up-subscription = { workspace = true }
 up-transport-mqtt5 = { version = "0.4.0", optional = true }
 up-transport-zenoh = { version = "0.9.0", optional = true }

up-subscription-cli/Cargo.toml

@@ -11,18 +11,36 @@
  * SPDX-License-Identifier: Apache-2.0
  ********************************************************************************/
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
+use backon::{ExponentialBuilder, Retryable};
 use tracing::info;
-use up_rust::UTransport;
+use up_rust::{UCode, UTransport};
 use up_transport_mqtt5::{Mqtt5Transport, Mqtt5TransportOptions};
 
 pub(crate) async fn get_mqtt5_transport(
     authority_name: &str,
     mqtt5_args: Mqtt5TransportOptions,
 ) -> Result<Arc<dyn UTransport>, Box<dyn std::error::Error>> {
     info!("Using MQTT 5 uProtocol transport");
-    Ok(Mqtt5Transport::new(mqtt5_args, authority_name)
+
+    let transport = Mqtt5Transport::new(mqtt5_args, authority_name)
         .await
-        .map(Arc::new)?)
+        .map(Arc::new)?;
+
+    (|| transport.connect())
+                    .retry(
+                        ExponentialBuilder::default().with_total_delay(Some(Duration::from_secs(10))),
+                    )
+                    .notify(|error, sleep_duration| {
+                        info!("Attempt to connect to MQTT broker failed [error: {error}], retrying in {sleep_duration:?}");
+                    })
+                    .when(|err| {
+                        // no need to keep retrying if authentication or permission is denied
+                        err.get_code() != UCode::UNAUTHENTICATED
+                            && err.get_code() != UCode::PERMISSION_DENIED
+                    })
+                    .await?;
+    info!("Connected to MQTT5 broker");
+    Ok(transport)
 }