Update to paho-mqtt 0.14.0

Author: sophokles73

Cargo.lock

@@ -29,7 +29,7 @@ default = ["cli"]
 cli = ["clap"]
 
 [dependencies]
-async-channel = { version = "1.9" }
+async-channel = { version = "2.0.0" }
 async-trait = { version = "0.1" }
 backon = { version = "1.6", default-features = false, features = ["tokio-sleep"] }
 bytes = { version = "1.11.1" }
@@ -45,22 +45,22 @@ clap = { version = "4.5", optional = true, default-features = false, features =
 ] }
 futures = { version = "0.3" }
 log = { version = "0.4" }
-paho-mqtt = { version = "0.13.3", features = ["vendored-ssl"] }
+paho-mqtt = { version = "0.14.0", features = ["vendored-ssl"] }
 protobuf = { version = "3.7.2" }
 slab = { version = "0.4.11" }
 tokio = { version = "1.45.1", default-features = false, features = ["rt", "rt-multi-thread", "sync"] }
 up-rust = { version = "0.9", default-features = false }
 
 [build-dependencies]
-testcontainers = { version = "0.27.1", default-features = false, features = ["blocking"] }
+testcontainers = { version = "0.27.2", default-features = false, features = ["blocking"] }
 
 [dev-dependencies]
 env_logger = { version = "0.11.8" }
 mockall = { version = "0.14.0" }
 serial_test = { version = "3.2.0" }
 tempfile = { version = "3.23.0" }
 test-case = { version = "3.3.1" }
-testcontainers = { version = "0.27.1", default-features = false }
+testcontainers = { version = "0.27.2", default-features = false }
 # https://rustsec.org/advisories/RUSTSEC-2026-0049
 rustls-webpki = { version = "0.103.10" }
 # https://rustsec.org/advisories/RUSTSEC-2026-0009

Cargo.toml

@@ -6,7 +6,6 @@ accepted = [
     "Apache-2.0",
     "BSD-2-Clause",
     "BSD-3-Clause",
-    "CDLA-Permissive-2.0",
     "EPL-2.0",
     "ISC",
     "MIT",

about.toml

@@ -19,7 +19,6 @@ allow = [
     "Apache-2.0",
     "BSD-2-Clause",
     "BSD-3-Clause",
-    "CDLA-Permissive-2.0",
     "EPL-2.0",
     "ISC",
     "MIT",
@@ -39,6 +38,4 @@ skip-tree = [
 
 [advisories]
 ignore = [
-    # https://rustsec.org/advisories/RUSTSEC-2026-0066
-    { id = "RUSTSEC-2026-0066", reason = "this affects the testcontainers crate which is used for running tests only" },
 ]

deny.toml

@@ -36,7 +36,6 @@ use async_channel::Receiver;
 use bytes::Bytes;
 #[cfg(feature = "cli")]
 use clap::{Args, ValueEnum};
-use futures::stream::StreamExt;
 use listener_registry::{RegisteredListeners, SubscriptionIdentifier};
 use log::{debug, trace};
 use mqtt_client::MqttClientOperations;
@@ -429,12 +428,12 @@ impl Mqtt5Transport {
     /// * `registered_listeners` - Map of topic filters to listeners.
     /// * `message_stream` - Stream of incoming MQTT PUBLISH packets.
     /// * `mqtt_client_operations` - The client to use for interacting with the MQTT broker.
-    fn create_cb_message_handler(&mut self, mut message_stream: Receiver<Option<Message>>) {
+    fn create_cb_message_handler(&mut self, message_stream: Receiver<Option<Message>>) {
         let cloned_client_operations = self.mqtt_client.clone();
         let cloned_registered_listeners = self.registered_listeners.clone();
         let cloned_message_mapper = self.message_mapper.clone();
         let handle = tokio::spawn(async move {
-            while let Some(msg_opt) = message_stream.next().await {
+            while let Ok(msg_opt) = message_stream.recv().await {
                 let Some(msg) = msg_opt else {
                     // None means that the connection is dropped.
                     debug!("Lost connection to MQTT broker");

src/lib.rs

@@ -23,7 +23,7 @@ use async_trait::async_trait;
 use backon::Retryable;
 #[cfg(feature = "cli")]
 use clap::Args;
-use log::{debug, trace};
+use log::{debug, error, trace};
 use up_rust::{UCode, UStatus};
 
 use crate::{listener_registry::SubscribedTopicProvider, SubscriptionIdentifier};
@@ -150,7 +150,7 @@ impl TryFrom<&MqttClientOptions> for paho_mqtt::ConnectOptions {
             connect_options_builder.user_name(v);
         }
         if let Some(v) = options.password.as_ref() {
-            connect_options_builder.password(v);
+            connect_options_builder.password(v.as_bytes());
         }
         Ok(connect_options_builder.finalize())
     }
@@ -221,6 +221,49 @@ impl TryFrom<&SslOptions> for paho_mqtt::SslOptions {
     }
 }
 
+fn ustatus_from_paho_error(paho_error: paho_mqtt::Error) -> UStatus {
+    match paho_error {
+        paho_mqtt::Error::Disconnected => {
+            UStatus::fail_with_code(UCode::UNAVAILABLE, "not connected to MQTT broker")
+        }
+        paho_mqtt::Error::TcpTlsConnectFailure => {
+            UStatus::fail_with_code(UCode::UNAVAILABLE, "failed to connect to MQTT broker")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::BadUserNameOrPassword, _) => {
+            UStatus::fail_with_code(UCode::UNAUTHENTICATED, "bad credentials")
+        }
+        // [impl->dsn~mqtt5-transport-authorization~1]
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::NotAuthorized, _) => {
+            UStatus::fail_with_code(UCode::PERMISSION_DENIED, "not authorized")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::ServerUnavailable, _) => {
+            UStatus::fail_with_code(UCode::UNAVAILABLE, "server not available")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::ServerBusy, _) => {
+            UStatus::fail_with_code(UCode::UNAVAILABLE, "server busy")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::BadAuthenticationMethod, _) => {
+            UStatus::fail_with_code(UCode::UNAUTHENTICATED, "bad authentication method")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::MessageRateTooHigh, _) => {
+            UStatus::fail_with_code(UCode::RESOURCE_EXHAUSTED, "message rate to high")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::QuotaExceeded, _) => {
+            UStatus::fail_with_code(UCode::RESOURCE_EXHAUSTED, "quota exceeded")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::ConnectionRateExceeded, _) => {
+            UStatus::fail_with_code(UCode::RESOURCE_EXHAUSTED, "connection rate exceeded")
+        }
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::MaximumConnectTime, _) => {
+            UStatus::fail_with_code(UCode::RESOURCE_EXHAUSTED, "maximum connect time exceeded")
+        }
+        _ => {
+            error!("paho error: {paho_error:?}");
+            UStatus::fail_with_code(UCode::INTERNAL, paho_error.to_string())
+        }
+    }
+}
+
 /// Basic operations that an MQTT client performs.
 #[cfg_attr(test, mockall::automock)]
 #[async_trait]
@@ -331,18 +374,6 @@ pub(crate) struct PahoBasedMqttClientOperations {
 }
 
 impl PahoBasedMqttClientOperations {
-    fn ustatus_from_paho_error(paho_error: &paho_mqtt::Error) -> UStatus {
-        match paho_error {
-            paho_mqtt::Error::Disconnected => {
-                UStatus::fail_with_code(UCode::UNAVAILABLE, "not connected to MQTT broker")
-            }
-            paho_mqtt::Error::TcpTlsConnectFailure => {
-                UStatus::fail_with_code(UCode::UNAVAILABLE, "failed to connect to MQTT broker")
-            }
-            _ => UStatus::fail_with_code(UCode::UNKNOWN, paho_error.to_string()),
-        }
-    }
-
     /// Creates new MQTT client.
     ///
     /// # Arguments
@@ -491,12 +522,7 @@ impl MqttClientOperations for PahoBasedMqttClientOperations {
                     Self::process_connack(user_data, response);
                 }
             })
-            .map_err(|e| {
-                UStatus::fail_with_code(
-                    UCode::INTERNAL,
-                    format!("Failed to connect to MQTT broker: {e:?}"),
-                )
-            })
+            .map_err(ustatus_from_paho_error)
     }
 
     fn is_connected(&self) -> bool {
@@ -607,7 +633,7 @@ impl MqttClientOperations for PahoBasedMqttClientOperations {
         self.inner_mqtt_client
             .publish(mqtt_message)
             .await
-            .map_err(|paho_error| Self::ustatus_from_paho_error(&paho_error))
+            .map_err(ustatus_from_paho_error)
     }
 
     async fn subscribe(&self, topic: &str, id: u16) -> Result<(), UStatus> {
@@ -634,22 +660,25 @@ impl MqttClientOperations for PahoBasedMqttClientOperations {
             // QOS 1 - Delivered and received at least once
             .subscribe_with_options(topic, paho_mqtt::QOS_1, None, subscription_properties)
             .await
-            .map_err(|paho_error| Self::ustatus_from_paho_error(&paho_error))
+            .map_err(ustatus_from_paho_error)
             .map(|_| ())
     }
 
     async fn unsubscribe(&self, topic: &str) -> Result<(), UStatus> {
         self.inner_mqtt_client
             .unsubscribe(topic)
             .await
-            .map_err(|paho_error| Self::ustatus_from_paho_error(&paho_error))
+            .map_err(ustatus_from_paho_error)
             .map(|_| ())
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use paho_mqtt::ConnectOptions;
+    use paho_mqtt::{ConnectOptions, Properties};
+    use up_rust::UCode;
+
+    use crate::mqtt_client::ustatus_from_paho_error;
 
     use super::MqttClientOptions;
 
@@ -668,4 +697,18 @@ mod tests {
         // because the ConnectOptions struct does not (yet) provide access to the CONNECT packet
         // properties
     }
+
+    #[test_case::test_case(
+        paho_mqtt::Error::TcpTlsConnectFailure => UCode::UNAVAILABLE;
+        "connect failure")]
+    #[test_case::test_case(
+        paho_mqtt::Error::Disconnected => UCode::UNAVAILABLE;
+        "disconnected")]
+    // [utest->dsn~mqtt5-transport-authorization~1]
+    #[test_case::test_case(
+        paho_mqtt::Error::ReasonCode(paho_mqtt::ReasonCode::NotAuthorized, Properties::new()) => UCode::PERMISSION_DENIED;
+        "not authorized")]
+    fn test_ustatus_from_paho_error(paho_error: paho_mqtt::Error) -> UCode {
+        ustatus_from_paho_error(paho_error).get_code()
+    }
 }