feat(mqtt): add MQTT v5 client5 module with per-message properties

CHANGELOG.md

@@ -48,6 +48,7 @@ remote_component = { name = "espressif/lan87xx", version = "1.*" }
 - TLS: Server-side APIs (`ServerConfig`, `negotiate_server`, …) are available again on ESP-IDF >= 5.3 with mbedTLS. `CONFIG_ESP_TLS_SERVER` was removed in v5.3, so the old `esp_idf_esp_tls_server` cfg was never set and those APIs were compiled out; gating now uses `esp_idf_esp_tls_server` (IDF ≤ 5.2) or `esp_idf_version_at_least_5_3_0` + `esp_idf_esp_tls_using_mbedtls`.
 
 ### Added
+- MQTT5: new `client5` module (`EspMqtt5Client`, `EspMqtt5Connection`) implementing the `embedded-svc` MQTT v5 client traits, with per-message PUBLISH / SUBSCRIBE / UNSUBSCRIBE property support (MQTT 5.0 §3.3.2.3, §3.8.2.1, §3.10.2.1). Behind the new opt-in `mqtt_protocol_v5` feature, and additionally gated on `CONFIG_MQTT_PROTOCOL_5=y`.
 - Compatibility with ESP-IDF V6.0, and some pre-release 6.0.x.
 - Added support for the Generic Ethernet PHY driver: particularly useful on ESP-IDF 6.0+ as it is built-in.
 - Added type-safe wrappers for the NimBLE low-resource-use BLE stack: GAP, GATT Server, GATT Client, L2CAP. See `examples/ble_*.rs`

Cargo.toml

@@ -21,11 +21,19 @@ harness = false
 [features]
 default = ["std", "binstart"]
 
-std = ["alloc", "log/std", "esp-idf-hal/std", "embedded-svc/std", "futures-io"]
+std = [
+    "alloc",
+    "log/std",
+    "esp-idf-hal/std",
+    "embedded-svc/std",
+    "futures-io",
+]
 embassy-time-driver = ["dep:embassy-time-driver", "embassy-time-queue-utils"]
 alloc = ["esp-idf-hal/alloc", "embedded-svc/alloc", "uncased/alloc"]
 nightly = ["embedded-svc/nightly", "esp-idf-hal/nightly"]
 experimental = ["embedded-svc/experimental", "esp-idf-hal/experimental"]
+# MQTT 5.0 client (`mqtt::client5`). Requires `CONFIG_MQTT_PROTOCOL_5=y` in sdkconfig.
+mqtt_protocol_v5 = ["std", "embedded-svc/mqtt_protocol_v5"]
 
 # Propagated esp-idf-hal features
 critical-section = ["esp-idf-hal/critical-section"]
@@ -76,3 +84,6 @@ async-io = { version = "0.4", package = "async-io-mini", default-features = fals
 [patch.crates-io]
 esp-idf-sys = { git = "https://github.com/esp-rs/esp-idf-sys", branch = "master" }
 esp-idf-hal = { git = "https://github.com/esp-rs/esp-idf-hal", branch = "master" }
+# Needed for the `mqtt_protocol_v5` feature (the `mqtt::client5` traits), which is not in
+# a published embedded-svc release yet. Drop once embedded-svc is released with it.
+embedded-svc = { git = "https://github.com/esp-rs/embedded-svc", branch = "master" }

src/mqtt.rs

@@ -3,3 +3,6 @@
 //! MQTT is a lightweight publish/subscribe messaging protocol.
 
 pub mod client;
+
+#[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+pub mod client5;

src/mqtt/client.rs

@@ -2,13 +2,24 @@
 use core::ffi::c_void;
 use core::fmt::Debug;
 use core::{slice, time};
+#[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+use std::vec::Vec;
 
 extern crate alloc;
 use alloc::boxed::Box;
 use alloc::sync::Arc;
 
 use embedded_svc::mqtt::client::{asynch, Client, Connection, Enqueue, ErrorType, Publish};
 
+#[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+use embedded_svc::mqtt::client5::{MessageMetadata, SubscribePropertyConfig, UserPropertyItem};
+
+#[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+use embedded_svc::mqtt::client5::UserPropertyList;
+
+#[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+use crate::mqtt::client5::EspUserPropertyList;
+
 use crate::private::unblocker::Unblocker;
 use crate::sys::*;
 
@@ -25,6 +36,16 @@ pub use embedded_svc::mqtt::client::{
 #[allow(unused_imports)]
 pub use super::*;
 
+fn u8ptr_to_str<'a>(ptr: *const u8, len: usize) -> Option<&'a str> {
+    if ptr.is_null() || len == 0 {
+        return None;
+    }
+
+    // SAFETY: The pointer is assumed to be valid and the length is non-zero.
+    let slice: &'a [u8] = unsafe { core::slice::from_raw_parts(ptr, len) };
+    core::str::from_utf8(slice).ok()
+}
+
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum MqttProtocolVersion {
@@ -373,22 +394,24 @@ impl<'a> TryFrom<&'a MqttClientConfiguration<'a>>
     }
 }
 
-struct UnsafeCallback<'a>(*mut Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>);
+pub(crate) struct UnsafeCallback<'a>(*mut Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>);
 
 impl<'a> UnsafeCallback<'a> {
-    fn from(boxed: &mut Box<Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>>) -> Self {
+    pub(crate) fn from(
+        boxed: &mut Box<Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>>,
+    ) -> Self {
         Self(boxed.as_mut())
     }
 
-    unsafe fn from_ptr(ptr: *mut c_void) -> Self {
+    pub(crate) unsafe fn from_ptr(ptr: *mut c_void) -> Self {
         Self(ptr as *mut _)
     }
 
-    fn as_ptr(&self) -> *mut c_void {
+    pub(crate) fn as_ptr(&self) -> *mut c_void {
         self.0 as *mut _
     }
 
-    unsafe fn call(&self, data: esp_mqtt_event_handle_t) {
+    pub(crate) unsafe fn call(&self, data: esp_mqtt_event_handle_t) {
         let reference = self.0.as_mut().unwrap();
 
         (reference)(data);
@@ -595,10 +618,29 @@ impl<'a> EspMqttClient<'a> {
         self.subscribe_cstr(to_cstring_arg(topic)?.as_c_str(), qos)
     }
 
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    pub fn subscribe_with_config<'ab>(
+        &mut self,
+        topic: &str,
+        qos: QoS,
+        config: SubscribePropertyConfig<'ab>,
+    ) -> Result<MessageId, EspError> {
+        self.subscribe_with_config_cstr(to_cstring_arg(topic)?.as_c_str(), qos, config)
+    }
+
     pub fn unsubscribe(&mut self, topic: &str) -> Result<MessageId, EspError> {
         self.unsubscribe_cstr(to_cstring_arg(topic)?.as_c_str())
     }
 
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    pub fn unsubscribe_with_config<'ab>(
+        &mut self,
+        topic: &str,
+        config: SubscribePropertyConfig<'ab>,
+    ) -> Result<MessageId, EspError> {
+        self.unsubscribe_with_config_cstr(to_cstring_arg(topic)?.as_c_str(), config)
+    }
+
     pub fn publish(
         &mut self,
         topic: &str,
@@ -653,10 +695,71 @@ impl<'a> EspMqttClient<'a> {
         res
     }
 
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    pub fn subscribe_with_config_cstr(
+        &mut self,
+        topic: &core::ffi::CStr,
+        qos: QoS,
+        config: SubscribePropertyConfig<'_>,
+    ) -> Result<MessageId, EspError> {
+        // `share_name` is a `&str`, which is not NUL-terminated: the C side runs
+        // `strlen()` on it, and `esp_mqtt5_client_set_subscribe_property` stores the
+        // pointer rather than copying. Keep an arena-owned CString alive across both
+        // the setter and the subscribe call below.
+        let mut cstrs = RawCstrs::new();
+
+        let property = esp_mqtt5_subscribe_property_config_t {
+            subscribe_id: config.subscribe_id,
+            no_local_flag: config.no_local,
+            retain_as_published_flag: config.retain_as_published,
+            retain_handle: config.retain_handling,
+            is_share_subscribe: config.share_name.is_some(),
+            share_name: cstrs.as_nptr(config.share_name)?,
+            user_property: if let Some(ref user_properties) = config.user_properties {
+                EspUserPropertyList::from(user_properties).as_ptr()
+            } else {
+                mqtt5_user_property_handle_t::default()
+            },
+        };
+
+        Self::check(unsafe {
+            esp_mqtt5_client_set_subscribe_property(self.raw_client, &property as *const _)
+        })?;
+
+        self.subscribe_cstr(topic, qos)
+    }
+
     pub fn unsubscribe_cstr(&mut self, topic: &core::ffi::CStr) -> Result<MessageId, EspError> {
         Self::check(unsafe { esp_mqtt_client_unsubscribe(self.raw_client, topic.as_ptr()) })
     }
 
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    pub fn unsubscribe_with_config_cstr<'ab>(
+        &mut self,
+        topic: &core::ffi::CStr,
+        config: SubscribePropertyConfig<'ab>,
+    ) -> Result<MessageId, EspError> {
+        // See `subscribe_with_config_cstr`: `&str` is not NUL-terminated and the C
+        // setter keeps the pointer, so the CString must outlive the unsubscribe call.
+        let mut cstrs = RawCstrs::new();
+
+        let property = esp_mqtt5_unsubscribe_property_config_t {
+            is_share_subscribe: config.share_name.is_some(),
+            share_name: cstrs.as_nptr(config.share_name)?,
+            user_property: if let Some(ref user_properties) = config.user_properties {
+                EspUserPropertyList::from(user_properties).as_ptr()
+            } else {
+                mqtt5_user_property_handle_t::default()
+            },
+        };
+
+        Self::check(unsafe {
+            esp_mqtt5_client_set_unsubscribe_property(self.raw_client, &property as *const _)
+        })?;
+
+        self.unsubscribe_cstr(topic)
+    }
+
     pub fn publish_cstr(
         &mut self,
         topic: &core::ffi::CStr,
@@ -751,7 +854,7 @@ impl ErrorType for EspMqttClient<'_> {
     type Error = EspError;
 }
 
-impl Client for EspMqttClient<'_> {
+impl<'a> Client for EspMqttClient<'a> {
     fn subscribe(&mut self, topic: &str, qos: QoS) -> Result<MessageId, Self::Error> {
         EspMqttClient::subscribe(self, topic, qos)
     }
@@ -788,8 +891,8 @@ impl Enqueue for EspMqttClient<'_> {
 unsafe impl Send for EspMqttClient<'_> {}
 
 pub struct EspMqttConnection {
-    receiver: Receiver<EspMqttEvent<'static>>,
-    given: bool,
+    pub(crate) receiver: Receiver<EspMqttEvent<'static>>,
+    pub(crate) given: bool,
 }
 
 impl EspMqttConnection {
@@ -1003,6 +1106,7 @@ impl EspAsyncMqttClient {
                 AsyncCommand::Subscribe { qos } => {
                     let topic =
                         unsafe { core::ffi::CStr::from_bytes_with_nul_unchecked(&work.topic) };
+
                     work.result = client.subscribe_cstr(topic, qos);
                 }
                 AsyncCommand::Unsubscribe => {
@@ -1091,7 +1195,7 @@ static ERROR: EspError = EspError::from_infallible::<ESP_FAIL>();
 pub struct EspMqttEvent<'a>(&'a esp_mqtt_event_t);
 
 impl<'a> EspMqttEvent<'a> {
-    const fn new(event: &'a esp_mqtt_event_t) -> Self {
+    pub(crate) const fn new(event: &'a esp_mqtt_event_t) -> Self {
         Self(event)
     }
 
@@ -1160,6 +1264,83 @@ impl<'a> EspMqttEvent<'a> {
             other => panic!("Unknown message type: {other}"),
         }
     }
+
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    pub fn metadata<'ab>(&self) -> Option<MessageMetadata<'ab>> {
+        let ptr = self.0.property;
+
+        if ptr.is_null() {
+            return None;
+        }
+
+        let payload_format_indicator = unsafe { (*ptr).payload_format_indicator };
+        let response_topic: Option<&'ab str> = unsafe {
+            let topic = (*ptr).response_topic;
+            let len = (*ptr).response_topic_len;
+            u8ptr_to_str(topic, len as _)
+        };
+
+        let correlation_data: Option<&'ab [u8]> = unsafe {
+            let data = (*ptr).correlation_data;
+            if data.is_null() {
+                None
+            } else {
+                Some(core::slice::from_raw_parts(
+                    data,
+                    (*ptr).correlation_data_len as usize,
+                ))
+            }
+        };
+
+        let content_type: Option<&'ab str> = unsafe {
+            let content_type = (*ptr).content_type;
+            let len = (*ptr).content_type_len;
+            u8ptr_to_str(content_type, len as _)
+        };
+
+        let subscribe_id = unsafe { (*ptr).subscribe_id };
+
+        let event_property = MessageMetadata::new(
+            payload_format_indicator,
+            response_topic,
+            correlation_data,
+            content_type,
+            subscribe_id,
+        );
+        Some(event_property)
+    }
+
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    fn user_properties<'ab>(&self) -> Result<Vec<UserPropertyItem<'ab>>, EspError> {
+        let count = self.user_properties_count();
+        if count == 0 {
+            return Ok(Vec::new());
+        }
+
+        let ptr = self.0.property;
+        if ptr.is_null() {
+            return Ok(Vec::new());
+        }
+        let table: *mut mqtt5_user_property_list_t = unsafe { (*ptr).user_property };
+        if table.is_null() {
+            return Ok(Vec::new());
+        }
+
+        Ok(EspUserPropertyList::from_handle(table).get_items()?)
+    }
+
+    fn user_properties_count(&self) -> u8 {
+        let ptr = self.0.property;
+        if ptr.is_null() {
+            return 0;
+        }
+        let table: *mut mqtt5_user_property_list_t = unsafe { (*ptr).user_property };
+        if table.is_null() {
+            return 0;
+        }
+        let table = EspUserPropertyList::from_handle(table);
+        table.count()
+    }
 }
 
 /// SAFETY: EspMqttEvent contains no thread-specific data.
@@ -1176,4 +1357,14 @@ impl Event for EspMqttEvent<'_> {
     fn payload(&self) -> EventPayload<'_, Self::Error> {
         EspMqttEvent::payload(self)
     }
+
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    fn metadata<'a>(&self) -> Option<MessageMetadata<'a>> {
+        EspMqttEvent::metadata(self)
+    }
+
+    #[cfg(all(esp_idf_mqtt_protocol_5, feature = "mqtt_protocol_v5"))]
+    fn user_properties<'ab>(&self) -> Result<Vec<UserPropertyItem<'ab>>, Self::Error> {
+        EspMqttEvent::user_properties(self)
+    }
 }