esp_idf_svc/mqtt/
client.rs

1//! MQTT protocol client
2use core::ffi::c_void;
3use core::fmt::Debug;
4use core::{slice, time};
5
6extern crate alloc;
7use alloc::boxed::Box;
8use alloc::sync::Arc;
9
10use embedded_svc::mqtt::client::{asynch, Client, Connection, Enqueue, ErrorType, Publish};
11
12use crate::private::unblocker::Unblocker;
13use crate::sys::*;
14
15use crate::handle::RawHandle;
16
17use crate::private::cstr::*;
18use crate::private::zerocopy::{Channel, QuitOnDrop, Receiver};
19use crate::tls::*;
20
21pub use embedded_svc::mqtt::client::{
22    Details, Event, EventPayload, InitialChunkData, MessageId, QoS, SubsequentChunkData,
23};
24
25#[allow(unused_imports)]
26pub use super::*;
27
28#[derive(Copy, Clone, Debug, Eq, PartialEq)]
29#[non_exhaustive]
30pub enum MqttProtocolVersion {
31    V3_1,
32    V3_1_1,
33    #[cfg(esp_idf_mqtt_protocol_5)]
34    V5,
35}
36
37impl From<MqttProtocolVersion> for esp_mqtt_protocol_ver_t {
38    fn from(pv: MqttProtocolVersion) -> Self {
39        match pv {
40            MqttProtocolVersion::V3_1 => esp_mqtt_protocol_ver_t_MQTT_PROTOCOL_V_3_1,
41            MqttProtocolVersion::V3_1_1 => esp_mqtt_protocol_ver_t_MQTT_PROTOCOL_V_3_1_1,
42            #[cfg(esp_idf_mqtt_protocol_5)]
43            MqttProtocolVersion::V5 => esp_mqtt_protocol_ver_t_MQTT_PROTOCOL_V_5,
44        }
45    }
46}
47
48/// MQTT 5.0 CONNECT properties. Requires `protocol_version: Some(MqttProtocolVersion::V5)`.
49/// `None` on a field leaves the property unset and the broker applies the MQTT 5 default.
50#[cfg(esp_idf_mqtt_protocol_5)]
51#[derive(Debug, Clone, Default)]
52pub struct Mqtt5ConnectionPropertyConfig {
53    /// Session Expiry Interval, seconds (MQTT5 §3.1.2.11.2).
54    pub session_expiry_interval: Option<u32>,
55    /// Will Delay Interval, seconds (MQTT5 §3.1.3.2.2).
56    pub will_delay_interval: Option<u32>,
57    /// Receive Maximum, max concurrent inbound QoS>0 PUBLISHes (MQTT5 §3.1.2.11.3).
58    pub receive_maximum: Option<u16>,
59    /// Maximum Packet Size the client accepts, bytes (MQTT5 §3.1.2.11.4).
60    pub maximum_packet_size: Option<u32>,
61    /// Topic Alias Maximum the broker may use (MQTT5 §3.1.2.11.5).
62    pub topic_alias_maximum: Option<u16>,
63    /// Request Response Information (MQTT5 §3.1.2.11.6). C field: `request_resp_info`.
64    pub request_response_info: Option<bool>,
65    /// Request Problem Information (MQTT5 §3.1.2.11.7). Protocol default is `true`.
66    pub request_problem_info: Option<bool>,
67    /// Will Message Expiry Interval, seconds (MQTT5 §3.3.2.3.3).
68    pub message_expiry_interval: Option<u32>,
69    /// Will Payload Format Indicator: UTF-8 (`true`) or bytes (`false`) (MQTT5 §3.3.2.3.2).
70    pub payload_format_indicator: Option<bool>,
71}
72
73#[derive(Debug)]
74pub struct LwtConfiguration<'a> {
75    pub topic: &'a str,
76    pub payload: &'a [u8],
77    pub qos: QoS,
78    pub retain: bool,
79}
80
81#[derive(Debug)]
82pub struct MqttClientConfiguration<'a> {
83    pub protocol_version: Option<MqttProtocolVersion>,
84
85    #[cfg(esp_idf_mqtt_protocol_5)]
86    pub mqtt5_connection_property: Option<Mqtt5ConnectionPropertyConfig>,
87
88    pub client_id: Option<&'a str>,
89
90    pub connection_refresh_interval: time::Duration,
91    pub keep_alive_interval: Option<time::Duration>,
92    pub reconnect_timeout: Option<time::Duration>,
93    pub network_timeout: time::Duration,
94
95    pub lwt: Option<LwtConfiguration<'a>>,
96
97    pub disable_clean_session: bool,
98
99    pub task_prio: u8,
100    pub task_stack: usize,
101    pub buffer_size: usize,
102    pub out_buffer_size: usize,
103    pub outbox_limit: Option<usize>,
104
105    pub username: Option<&'a str>,
106    pub password: Option<&'a str>,
107
108    pub use_global_ca_store: bool,
109    pub skip_cert_common_name_check: bool,
110    pub crt_bundle_attach: Option<unsafe extern "C" fn(conf: *mut c_void) -> esp_err_t>,
111
112    pub server_certificate: Option<X509<'static>>,
113
114    pub client_certificate: Option<X509<'static>>,
115    pub private_key: Option<X509<'static>>,
116    pub private_key_password: Option<&'a str>,
117
118    #[cfg(all(esp_idf_esp_tls_psk_verification, feature = "alloc"))]
119    pub psk: Option<Psk<'a>>,
120    // pub alpn_protos: &'a [&'a str],
121    // pub use_secure_element: bool,
122    // void *ds_data;                          /*!< carrier of handle for digital signature parameters */
123}
124
125impl Default for MqttClientConfiguration<'_> {
126    fn default() -> Self {
127        Self {
128            protocol_version: None,
129
130            #[cfg(esp_idf_mqtt_protocol_5)]
131            mqtt5_connection_property: None,
132
133            client_id: None,
134
135            connection_refresh_interval: time::Duration::from_secs(0),
136            keep_alive_interval: Some(time::Duration::from_secs(0)),
137            reconnect_timeout: Some(time::Duration::from_secs(0)),
138            network_timeout: time::Duration::from_secs(0),
139
140            lwt: None,
141
142            disable_clean_session: false,
143
144            task_prio: 0,
145            task_stack: 0,
146            buffer_size: 0,
147            out_buffer_size: 0,
148            outbox_limit: None,
149
150            username: None,
151            password: None,
152
153            use_global_ca_store: false,
154            skip_cert_common_name_check: false,
155
156            crt_bundle_attach: Default::default(),
157
158            server_certificate: None,
159
160            client_certificate: None,
161            private_key: None,
162            private_key_password: None,
163
164            #[cfg(all(esp_idf_esp_tls_psk_verification, feature = "alloc"))]
165            psk: None,
166        }
167    }
168}
169
170#[cfg(esp_idf_version_major = "4")]
171impl<'a> TryFrom<&'a MqttClientConfiguration<'a>>
172    for (esp_mqtt_client_config_t, RawCstrs, Option<TlsPsk>)
173{
174    type Error = EspError;
175
176    fn try_from(conf: &'a MqttClientConfiguration<'a>) -> Result<Self, Self::Error> {
177        let mut cstrs = RawCstrs::new();
178
179        let mut c_conf = esp_mqtt_client_config_t {
180            protocol_ver: if let Some(protocol_version) = conf.protocol_version {
181                protocol_version.into()
182            } else {
183                esp_mqtt_protocol_ver_t_MQTT_PROTOCOL_UNDEFINED
184            },
185            client_id: cstrs.as_nptr(conf.client_id)?,
186
187            refresh_connection_after_ms: conf.connection_refresh_interval.as_millis() as _,
188            network_timeout_ms: conf.network_timeout.as_millis() as _,
189
190            disable_clean_session: conf.disable_clean_session as _,
191
192            task_prio: conf.task_prio as _,
193            task_stack: conf.task_stack as _,
194            buffer_size: conf.buffer_size as _,
195            out_buffer_size: conf.out_buffer_size as _,
196
197            username: cstrs.as_nptr(conf.username)?,
198            password: cstrs.as_nptr(conf.password)?,
199
200            use_global_ca_store: conf.use_global_ca_store,
201            skip_cert_common_name_check: conf.skip_cert_common_name_check,
202            crt_bundle_attach: conf.crt_bundle_attach,
203
204            ..Default::default()
205        };
206
207        if let Some(keep_alive_interval) = conf.keep_alive_interval {
208            c_conf.keepalive = keep_alive_interval.as_secs() as _;
209            c_conf.disable_keepalive = false;
210        } else {
211            c_conf.disable_keepalive = true;
212        }
213
214        if let Some(reconnect_timeout) = conf.reconnect_timeout {
215            c_conf.reconnect_timeout_ms = reconnect_timeout.as_millis() as _;
216            c_conf.disable_auto_reconnect = false;
217        } else {
218            c_conf.disable_auto_reconnect = true;
219        }
220
221        if let Some(lwt) = conf.lwt.as_ref() {
222            c_conf.lwt_topic = cstrs.as_ptr(lwt.topic)?;
223            c_conf.lwt_msg = if lwt.payload.is_empty() {
224                // ... or else len = 0 will cause ESP-IDF to treat our pointer as
225                // if it points to a C string, which it definitely doesn't
226                core::ptr::null()
227            } else {
228                lwt.payload.as_ptr() as _
229            };
230            c_conf.lwt_msg_len = lwt.payload.len() as _;
231            c_conf.lwt_qos = lwt.qos as _;
232            c_conf.lwt_retain = lwt.retain as _;
233        }
234
235        if let Some(cert) = conf.server_certificate {
236            c_conf.cert_pem = cert.as_esp_idf_raw_ptr() as _;
237            c_conf.cert_len = cert.as_esp_idf_raw_len();
238        }
239
240        if let (Some(cert), Some(private_key)) = (conf.client_certificate, conf.private_key) {
241            c_conf.client_cert_pem = cert.as_esp_idf_raw_ptr() as _;
242            c_conf.client_cert_len = cert.as_esp_idf_raw_len();
243
244            c_conf.client_key_pem = private_key.as_esp_idf_raw_ptr() as _;
245            c_conf.client_key_len = private_key.as_esp_idf_raw_len();
246
247            if let Some(pass) = conf.private_key_password {
248                c_conf.clientkey_password = pass.as_ptr() as _;
249                c_conf.clientkey_password_len = pass.len() as _;
250            }
251        }
252
253        #[cfg(all(esp_idf_esp_tls_psk_verification, feature = "alloc"))]
254        let tls_psk_conf = conf.psk.as_ref().map(|psk| psk.try_into()).transpose()?;
255        #[cfg(not(all(esp_idf_esp_tls_psk_verification, feature = "alloc")))]
256        let tls_psk_conf = None;
257
258        Ok((c_conf, cstrs, tls_psk_conf))
259    }
260}
261
262#[allow(clippy::needless_update)]
263#[cfg(not(esp_idf_version_major = "4"))]
264impl<'a> TryFrom<&'a MqttClientConfiguration<'a>>
265    for (esp_mqtt_client_config_t, RawCstrs, Option<TlsPsk>)
266{
267    type Error = EspError;
268
269    fn try_from(conf: &'a MqttClientConfiguration<'a>) -> Result<Self, EspError> {
270        let mut cstrs = RawCstrs::new();
271
272        #[allow(clippy::needless_update)]
273        let mut c_conf = esp_mqtt_client_config_t {
274            broker: esp_mqtt_client_config_t_broker_t {
275                verification: esp_mqtt_client_config_t_broker_t_verification_t {
276                    use_global_ca_store: conf.use_global_ca_store,
277                    skip_cert_common_name_check: conf.skip_cert_common_name_check,
278                    crt_bundle_attach: conf.crt_bundle_attach,
279                    ..Default::default()
280                },
281                ..Default::default()
282            },
283            credentials: esp_mqtt_client_config_t_credentials_t {
284                client_id: cstrs.as_nptr(conf.client_id)?,
285                set_null_client_id: conf.client_id.is_none(),
286                username: cstrs.as_nptr(conf.username)?,
287                authentication: esp_mqtt_client_config_t_credentials_t_authentication_t {
288                    password: cstrs.as_nptr(conf.password)?,
289                    ..Default::default()
290                },
291                ..Default::default()
292            },
293            session: esp_mqtt_client_config_t_session_t {
294                protocol_ver: if let Some(protocol_version) = conf.protocol_version {
295                    protocol_version.into()
296                } else {
297                    esp_mqtt_protocol_ver_t_MQTT_PROTOCOL_UNDEFINED
298                },
299                disable_clean_session: conf.disable_clean_session as _,
300                ..Default::default()
301            },
302            network: esp_mqtt_client_config_t_network_t {
303                refresh_connection_after_ms: conf.connection_refresh_interval.as_millis() as _,
304                timeout_ms: conf.network_timeout.as_millis() as _,
305                ..Default::default()
306            },
307            task: esp_mqtt_client_config_t_task_t {
308                priority: conf.task_prio as _,
309                stack_size: conf.task_stack as _,
310                ..Default::default()
311            },
312            buffer: esp_mqtt_client_config_t_buffer_t {
313                size: conf.buffer_size as _,
314                out_size: conf.out_buffer_size as _,
315                ..Default::default()
316            },
317            ..Default::default()
318        };
319
320        if let Some(keep_alive_interval) = conf.keep_alive_interval {
321            c_conf.session.keepalive = keep_alive_interval.as_secs() as _;
322            c_conf.session.disable_keepalive = false;
323        } else {
324            c_conf.session.disable_keepalive = true;
325        }
326
327        if let Some(reconnect_timeout) = conf.reconnect_timeout {
328            c_conf.network.reconnect_timeout_ms = reconnect_timeout.as_millis() as _;
329            c_conf.network.disable_auto_reconnect = false;
330        } else {
331            c_conf.network.disable_auto_reconnect = true;
332        }
333
334        if let Some(lwt) = conf.lwt.as_ref() {
335            c_conf.session.last_will = esp_mqtt_client_config_t_session_t_last_will_t {
336                topic: cstrs.as_ptr(lwt.topic)?,
337                msg: lwt.payload.as_ptr() as _,
338                msg_len: lwt.payload.len() as _,
339                qos: lwt.qos as _,
340                retain: lwt.retain as _,
341                ..Default::default()
342            };
343        }
344
345        if let Some(cert) = conf.server_certificate {
346            c_conf.broker.verification.certificate = cert.as_esp_idf_raw_ptr() as _;
347            c_conf.broker.verification.certificate_len = cert.as_esp_idf_raw_len();
348        }
349
350        if let (Some(cert), Some(private_key)) = (conf.client_certificate, conf.private_key) {
351            c_conf.credentials.authentication.certificate = cert.as_esp_idf_raw_ptr() as _;
352            c_conf.credentials.authentication.certificate_len = cert.as_esp_idf_raw_len();
353
354            c_conf.credentials.authentication.key = private_key.as_esp_idf_raw_ptr() as _;
355            c_conf.credentials.authentication.key_len = private_key.as_esp_idf_raw_len();
356
357            if let Some(pass) = conf.private_key_password {
358                c_conf.credentials.authentication.key_password = pass.as_ptr() as _;
359                c_conf.credentials.authentication.key_password_len = pass.len() as _;
360            }
361        }
362
363        if let Some(outbox_limit) = conf.outbox_limit {
364            c_conf.outbox.limit = outbox_limit as _;
365        }
366
367        #[cfg(all(esp_idf_esp_tls_psk_verification, feature = "alloc"))]
368        let tls_psk_conf = conf.psk.as_ref().map(|psk| psk.try_into()).transpose()?;
369        #[cfg(not(all(esp_idf_esp_tls_psk_verification, feature = "alloc")))]
370        let tls_psk_conf = None;
371
372        Ok((c_conf, cstrs, tls_psk_conf))
373    }
374}
375
376struct UnsafeCallback<'a>(*mut Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>);
377
378impl<'a> UnsafeCallback<'a> {
379    fn from(boxed: &mut Box<Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>>) -> Self {
380        Self(boxed.as_mut())
381    }
382
383    unsafe fn from_ptr(ptr: *mut c_void) -> Self {
384        Self(ptr as *mut _)
385    }
386
387    fn as_ptr(&self) -> *mut c_void {
388        self.0 as *mut _
389    }
390
391    unsafe fn call(&self, data: esp_mqtt_event_handle_t) {
392        let reference = self.0.as_mut().unwrap();
393
394        (reference)(data);
395    }
396}
397
398pub struct EspMqttClient<'a> {
399    raw_client: esp_mqtt_client_handle_t,
400    _boxed_raw_callback: Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>,
401    _tls_psk_conf: Option<TlsPsk>,
402}
403
404impl RawHandle for EspMqttClient<'_> {
405    type Handle = esp_mqtt_client_handle_t;
406
407    fn handle(&self) -> Self::Handle {
408        self.raw_client
409    }
410}
411
412impl EspMqttClient<'static> {
413    pub fn new(
414        url: &str,
415        conf: &MqttClientConfiguration,
416    ) -> Result<(Self, EspMqttConnection), EspError>
417    where
418        Self: Sized,
419    {
420        let (channel, receiver) = Channel::new();
421
422        let sender = QuitOnDrop::new(channel);
423
424        let conn = EspMqttConnection {
425            receiver,
426            given: false,
427        };
428
429        let client = Self::new_cb(url, conf, move |mut event| {
430            let event: &mut EspMqttEvent<'static> = unsafe { core::mem::transmute(&mut event) };
431            sender.channel().share(event);
432        })?;
433
434        Ok((client, conn))
435    }
436
437    pub fn new_cb<F>(
438        url: &str,
439        conf: &MqttClientConfiguration,
440        callback: F,
441    ) -> Result<Self, EspError>
442    where
443        F: for<'b> FnMut(EspMqttEvent<'b>) + Send + 'static,
444        Self: Sized,
445    {
446        unsafe { Self::new_nonstatic_cb(url, conf, callback) }
447    }
448}
449
450impl<'a> EspMqttClient<'a> {
451    /// # Safety
452    ///
453    /// This method - in contrast to method `new_generic` - allows the user to pass
454    /// a non-static callback/closure. This enables users to borrow
455    /// - in the closure - variables that live on the stack - or more generally - in the same
456    ///   scope where the service is created.
457    ///
458    /// HOWEVER: care should be taken NOT to call `core::mem::forget()` on the service,
459    /// as that would immediately lead to an UB (crash).
460    /// Also note that forgetting the service might happen with `Rc` and `Arc`
461    /// when circular references are introduced: <https://github.com/rust-lang/rust/issues/24456>
462    ///
463    /// The reason is that the closure is actually sent to a hidden ESP IDF thread.
464    /// This means that if the service is forgotten, Rust is free to e.g. unwind the stack
465    /// and the closure now owned by this other thread will end up with references to variables that no longer exist.
466    ///
467    /// The destructor of the service takes care - prior to the service being dropped and e.g.
468    /// the stack being unwind - to remove the closure from the hidden thread and destroy it.
469    /// Unfortunately, when the service is forgotten, the un-subscription does not happen
470    /// and invalid references are left dangling.
471    ///
472    /// This "local borrowing" will only be possible to express in a safe way once/if `!Leak` types
473    /// are introduced to Rust (i.e. the impossibility to "forget" a type and thus not call its destructor).
474    pub unsafe fn new_nonstatic_cb<F>(
475        url: &str,
476        conf: &MqttClientConfiguration,
477        mut callback: F,
478    ) -> Result<Self, EspError>
479    where
480        F: for<'b> FnMut(EspMqttEvent<'b>) + Send + 'a,
481        Self: Sized,
482    {
483        Self::new_raw(
484            url,
485            conf,
486            Box::new(move |event_handle| {
487                callback(EspMqttEvent::new(unsafe { event_handle.as_ref() }.unwrap()));
488            }),
489        )
490    }
491
492    fn new_raw(
493        url: &str,
494        conf: &MqttClientConfiguration,
495        raw_callback: Box<dyn FnMut(esp_mqtt_event_handle_t) + Send + 'a>,
496    ) -> Result<Self, EspError>
497    where
498        Self: Sized,
499    {
500        let mut boxed_raw_callback = Box::new(raw_callback);
501
502        let unsafe_callback = UnsafeCallback::from(&mut boxed_raw_callback);
503
504        let (mut c_conf, mut cstrs, tls_psk_conf) = conf.try_into()?;
505
506        #[cfg(esp_idf_version_major = "4")]
507        {
508            c_conf.uri = cstrs.as_ptr(url)?;
509        }
510
511        #[cfg(not(esp_idf_version_major = "4"))]
512        {
513            c_conf.broker.address.uri = cstrs.as_ptr(url)?;
514        }
515
516        #[cfg(all(esp_idf_esp_tls_psk_verification, feature = "alloc"))]
517        {
518            #[cfg(esp_idf_version_major = "4")]
519            if let Some(ref conf) = tls_psk_conf {
520                c_conf.psk_hint_key = &*conf.psk;
521            }
522            #[cfg(not(esp_idf_version_major = "4"))]
523            if let Some(ref conf) = tls_psk_conf {
524                c_conf.broker.verification.psk_hint_key = &*conf.psk;
525            }
526        }
527
528        let raw_client = unsafe { esp_mqtt_client_init(&c_conf as *const _) };
529        if raw_client.is_null() {
530            return Err(EspError::from_infallible::<ESP_FAIL>());
531        }
532
533        let client = Self {
534            raw_client,
535            _boxed_raw_callback: boxed_raw_callback,
536            _tls_psk_conf: tls_psk_conf,
537        };
538
539        esp!(unsafe {
540            esp_mqtt_client_register_event(
541                client.raw_client,
542                esp_mqtt_event_id_t_MQTT_EVENT_ANY,
543                Some(Self::handle),
544                unsafe_callback.as_ptr(),
545            )
546        })?;
547
548        // MQTT 5.0 CONNECT properties must be set after init and before start,
549        // otherwise the C call deadlocks on MQTT_API_LOCK held by mqtt_task.
550        #[cfg(esp_idf_mqtt_protocol_5)]
551        if let Some(props) = conf.mqtt5_connection_property.as_ref() {
552            if conf.protocol_version != Some(MqttProtocolVersion::V5) {
553                ::log::error!(
554                    "mqtt5_connection_property requires protocol_version = Some(MqttProtocolVersion::V5)"
555                );
556                return Err(EspError::from_infallible::<ESP_ERR_INVALID_ARG>());
557            }
558            let mut c_props = esp_mqtt5_connection_property_config_t::default();
559            if let Some(v) = props.session_expiry_interval {
560                c_props.session_expiry_interval = v;
561            }
562            if let Some(v) = props.will_delay_interval {
563                c_props.will_delay_interval = v;
564            }
565            if let Some(v) = props.receive_maximum {
566                c_props.receive_maximum = v;
567            }
568            if let Some(v) = props.maximum_packet_size {
569                c_props.maximum_packet_size = v;
570            }
571            if let Some(v) = props.topic_alias_maximum {
572                c_props.topic_alias_maximum = v;
573            }
574            if let Some(v) = props.request_response_info {
575                c_props.request_resp_info = v;
576            }
577            if let Some(v) = props.request_problem_info {
578                c_props.request_problem_info = v;
579            }
580            if let Some(v) = props.message_expiry_interval {
581                c_props.message_expiry_interval = v;
582            }
583            if let Some(v) = props.payload_format_indicator {
584                c_props.payload_format_indicator = v;
585            }
586            esp!(unsafe { esp_mqtt5_client_set_connect_property(client.raw_client, &c_props) })?;
587        }
588
589        esp!(unsafe { esp_mqtt_client_start(client.raw_client) })?;
590
591        Ok(client)
592    }
593
594    pub fn subscribe(&mut self, topic: &str, qos: QoS) -> Result<MessageId, EspError> {
595        self.subscribe_cstr(to_cstring_arg(topic)?.as_c_str(), qos)
596    }
597
598    pub fn unsubscribe(&mut self, topic: &str) -> Result<MessageId, EspError> {
599        self.unsubscribe_cstr(to_cstring_arg(topic)?.as_c_str())
600    }
601
602    pub fn publish(
603        &mut self,
604        topic: &str,
605        qos: QoS,
606        retain: bool,
607        payload: &[u8],
608    ) -> Result<MessageId, EspError> {
609        self.publish_cstr(to_cstring_arg(topic)?.as_c_str(), qos, retain, payload)
610    }
611
612    pub fn enqueue(
613        &mut self,
614        topic: &str,
615        qos: QoS,
616        retain: bool,
617        payload: &[u8],
618    ) -> Result<MessageId, EspError> {
619        self.enqueue_cstr(to_cstring_arg(topic)?.as_c_str(), qos, retain, payload)
620    }
621
622    pub fn subscribe_cstr(
623        &mut self,
624        topic: &core::ffi::CStr,
625        qos: QoS,
626    ) -> Result<MessageId, EspError> {
627        #[cfg(any(
628            esp_idf_version_major = "4",
629            all(esp_idf_version_major = "5", esp_idf_version_minor = "0"),
630            all(
631                esp_idf_version_major = "5",
632                esp_idf_version_minor = "1",
633                any(esp_idf_version_patch = "0", esp_idf_version_patch = "1")
634            )
635        ))]
636        let res = Self::check(unsafe {
637            esp_mqtt_client_subscribe(self.raw_client, topic.as_ptr(), qos as _)
638        });
639
640        #[cfg(not(any(
641            esp_idf_version_major = "4",
642            all(esp_idf_version_major = "5", esp_idf_version_minor = "0"),
643            all(
644                esp_idf_version_major = "5",
645                esp_idf_version_minor = "1",
646                any(esp_idf_version_patch = "0", esp_idf_version_patch = "1")
647            )
648        )))]
649        let res = Self::check(unsafe {
650            esp_mqtt_client_subscribe_single(self.raw_client, topic.as_ptr(), qos as _)
651        });
652
653        res
654    }
655
656    pub fn unsubscribe_cstr(&mut self, topic: &core::ffi::CStr) -> Result<MessageId, EspError> {
657        Self::check(unsafe { esp_mqtt_client_unsubscribe(self.raw_client, topic.as_ptr()) })
658    }
659
660    pub fn publish_cstr(
661        &mut self,
662        topic: &core::ffi::CStr,
663        qos: QoS,
664        retain: bool,
665        payload: &[u8],
666    ) -> Result<MessageId, EspError> {
667        let payload_ptr = match payload.len() {
668            0 => core::ptr::null(),
669            _ => payload.as_ptr(),
670        };
671
672        Self::check(unsafe {
673            esp_mqtt_client_publish(
674                self.raw_client,
675                topic.as_ptr(),
676                payload_ptr as _,
677                payload.len() as _,
678                qos as _,
679                retain as _,
680            )
681        })
682    }
683
684    pub fn enqueue_cstr(
685        &mut self,
686        topic: &core::ffi::CStr,
687        qos: QoS,
688        retain: bool,
689        payload: &[u8],
690    ) -> Result<MessageId, EspError> {
691        let payload_ptr = match payload.len() {
692            0 => core::ptr::null(),
693            _ => payload.as_ptr(),
694        };
695
696        Self::check(unsafe {
697            esp_mqtt_client_enqueue(
698                self.raw_client,
699                topic.as_ptr(),
700                payload_ptr as _,
701                payload.len() as _,
702                qos as _,
703                retain as _,
704                true,
705            )
706        })
707    }
708
709    pub fn set_uri(&mut self, uri: &str) -> Result<MessageId, EspError> {
710        self.set_uri_cstr(to_cstring_arg(uri)?.as_c_str())
711    }
712
713    pub fn set_uri_cstr(&mut self, uri: &core::ffi::CStr) -> Result<MessageId, EspError> {
714        Self::check(unsafe { esp_mqtt_client_set_uri(self.raw_client, uri.as_ptr()) })
715    }
716
717    pub fn get_outbox_size(&self) -> usize {
718        // this is always positive as internally this is converting uint64_t to int (defaults to 0)
719        let outbox_size = unsafe { esp_mqtt_client_get_outbox_size(self.raw_client) };
720        outbox_size.max(0) as usize
721    }
722
723    extern "C" fn handle(
724        event_handler_arg: *mut c_void,
725        _event_base: esp_event_base_t,
726        _event_id: i32,
727        event_data: *mut c_void,
728    ) {
729        unsafe {
730            UnsafeCallback::from_ptr(event_handler_arg).call(event_data as _);
731        }
732    }
733
734    fn check(result: i32) -> Result<MessageId, EspError> {
735        match EspError::from(result) {
736            Some(err) if result < 0 => Err(err),
737            _ => Ok(result as _),
738        }
739    }
740}
741
742impl Drop for EspMqttClient<'_> {
743    fn drop(&mut self) {
744        unsafe {
745            esp_mqtt_client_destroy(self.raw_client as _);
746        }
747    }
748}
749
750impl ErrorType for EspMqttClient<'_> {
751    type Error = EspError;
752}
753
754impl Client for EspMqttClient<'_> {
755    fn subscribe(&mut self, topic: &str, qos: QoS) -> Result<MessageId, Self::Error> {
756        EspMqttClient::subscribe(self, topic, qos)
757    }
758
759    fn unsubscribe(&mut self, topic: &str) -> Result<MessageId, Self::Error> {
760        EspMqttClient::unsubscribe(self, topic)
761    }
762}
763
764impl Publish for EspMqttClient<'_> {
765    fn publish(
766        &mut self,
767        topic: &str,
768        qos: QoS,
769        retain: bool,
770        payload: &[u8],
771    ) -> Result<MessageId, Self::Error> {
772        EspMqttClient::publish(self, topic, qos, retain, payload)
773    }
774}
775
776impl Enqueue for EspMqttClient<'_> {
777    fn enqueue(
778        &mut self,
779        topic: &str,
780        qos: QoS,
781        retain: bool,
782        payload: &[u8],
783    ) -> Result<MessageId, Self::Error> {
784        EspMqttClient::enqueue(self, topic, qos, retain, payload)
785    }
786}
787
788unsafe impl Send for EspMqttClient<'_> {}
789
790pub struct EspMqttConnection {
791    receiver: Receiver<EspMqttEvent<'static>>,
792    given: bool,
793}
794
795impl EspMqttConnection {
796    #[allow(clippy::should_implement_trait)]
797    pub fn next(&mut self) -> Result<&EspMqttEvent<'_>, EspError> {
798        if self.given {
799            self.receiver.done();
800        }
801
802        if let Some(event) = self.receiver.get_shared() {
803            self.given = true;
804
805            Ok(event)
806        } else {
807            self.given = false;
808
809            Err(EspError::from_infallible::<ESP_ERR_INVALID_STATE>())
810        }
811    }
812}
813
814impl ErrorType for EspMqttConnection {
815    type Error = EspError;
816}
817
818impl Connection for EspMqttConnection {
819    type Event<'a> = &'a EspMqttEvent<'a>;
820
821    fn next(&mut self) -> Result<Self::Event<'_>, Self::Error> {
822        EspMqttConnection::next(self)
823    }
824}
825
826#[derive(Copy, Clone, Debug)]
827enum AsyncCommand {
828    Subscribe { qos: QoS },
829    Unsubscribe,
830    Publish { qos: QoS, retain: bool },
831    SetUri,
832    None,
833}
834
835#[derive(Debug)]
836struct AsyncWork {
837    command: AsyncCommand,
838    topic: alloc::vec::Vec<u8>,
839    payload: alloc::vec::Vec<u8>,
840    result: Result<MessageId, EspError>,
841    broker_uri: alloc::vec::Vec<u8>,
842}
843
844pub struct EspAsyncMqttClient(Unblocker<AsyncWork>);
845
846impl EspAsyncMqttClient {
847    /// Create a new MQTT client with a given URL and configuration.
848    pub fn new(
849        url: &str,
850        conf: &MqttClientConfiguration<'_>,
851    ) -> Result<(Self, EspAsyncMqttConnection), EspError> {
852        Self::new_with_caps(url, conf, None)
853    }
854
855    /// Create a new MQTT client with a given URL, configuration
856    /// and caps.
857    ///
858    /// The caps tuple contains the initial capacity of the topic,
859    /// payload and broker URI buffers.
860    /// Useful to avoid constant re-allocations with large payloads.
861    pub fn new_with_caps(
862        url: &str,
863        conf: &MqttClientConfiguration<'_>,
864        caps: Option<(usize, usize, usize)>,
865    ) -> Result<(Self, EspAsyncMqttConnection), EspError> {
866        let (channel, receiver) = Channel::new();
867        let conn = EspAsyncMqttConnection {
868            receiver,
869            given: false,
870        };
871
872        let client = Self::wrap_with_caps(
873            EspMqttClient::new_cb(url, conf, move |mut event| {
874                let event: &mut EspMqttEvent<'static> = unsafe { core::mem::transmute(&mut event) };
875                channel.share(event);
876            })?,
877            caps,
878        )?;
879
880        Ok((client, conn))
881    }
882
883    /// Wrap an existing MQTT client with an async interface.
884    pub fn wrap(client: EspMqttClient<'static>) -> Result<Self, EspError> {
885        Self::wrap_with_caps(client, None)
886    }
887
888    /// Wrap an existing MQTT client with an async interface.
889    ///
890    /// The caps tuple contains the initial capacity of the topic,
891    /// payload and broker URI buffers.
892    /// Useful to avoid constant re-allocations with large payloads.
893    pub fn wrap_with_caps(
894        client: EspMqttClient<'static>,
895        caps: Option<(usize, usize, usize)>,
896    ) -> Result<Self, EspError> {
897        let unblocker = Unblocker::new(
898            CStr::from_bytes_until_nul(b"MQTT Sending task\0").unwrap(),
899            4096,
900            None,
901            None,
902            move |channel| Self::work(channel, client, caps),
903        )?;
904
905        Ok(Self(unblocker))
906    }
907
908    pub async fn subscribe(&mut self, topic: &str, qos: QoS) -> Result<MessageId, EspError> {
909        self.execute(AsyncCommand::Subscribe { qos }, Some(topic), None, None)
910            .await
911    }
912
913    pub async fn unsubscribe(&mut self, topic: &str) -> Result<MessageId, EspError> {
914        self.execute(AsyncCommand::Unsubscribe, Some(topic), None, None)
915            .await
916    }
917
918    pub async fn publish(
919        &mut self,
920        topic: &str,
921        qos: QoS,
922        retain: bool,
923        payload: &[u8],
924    ) -> Result<MessageId, EspError> {
925        self.execute(
926            AsyncCommand::Publish { qos, retain },
927            Some(topic),
928            Some(payload),
929            None,
930        )
931        .await
932    }
933
934    pub async fn set_uri(&mut self, broker_uri: &str) -> Result<MessageId, EspError> {
935        self.execute(AsyncCommand::SetUri, None, None, Some(broker_uri))
936            .await
937    }
938
939    async fn execute(
940        &mut self,
941        command: AsyncCommand,
942        topic: Option<&str>,
943        payload: Option<&[u8]>,
944        broker_uri: Option<&str>,
945    ) -> Result<MessageId, EspError> {
946        // Get the shared reference to the work item (as processed by the Self::work thread),
947        // and replace it with the next work item we want to process.
948        let work = self.0.exec_in_out().await.unwrap();
949
950        work.command = command;
951
952        if let Some(topic) = topic {
953            work.topic.clear();
954            work.topic.extend_from_slice(topic.as_bytes());
955            work.topic.push(0);
956        }
957
958        if let Some(payload) = payload {
959            work.payload.clear();
960            work.payload.extend_from_slice(payload);
961        }
962
963        if let Some(broker_uri) = broker_uri {
964            work.broker_uri.clear();
965            work.broker_uri.extend_from_slice(broker_uri.as_bytes());
966            work.broker_uri.push(0);
967        }
968
969        // Signal the worker thread that it can process the work item.
970        self.0.do_exec().await;
971
972        // Wait for the worker thread to finish and return the result.
973        let work = self.0.exec_in_out().await.unwrap();
974
975        work.result
976    }
977
978    fn work(
979        channel: Arc<Channel<AsyncWork>>,
980        mut client: EspMqttClient,
981        caps: Option<(usize, usize, usize)>,
982    ) {
983        // Placeholder work item. This will be replaced by the first actual work item.
984        let mut work = AsyncWork {
985            command: AsyncCommand::None,
986            topic: caps
987                .map(|cap| alloc::vec::Vec::with_capacity(cap.1))
988                .unwrap_or_default(),
989            payload: caps
990                .map(|cap| alloc::vec::Vec::with_capacity(cap.2))
991                .unwrap_or_default(),
992            result: Ok(0),
993            broker_uri: caps
994                .map(|cap| alloc::vec::Vec::with_capacity(cap.0))
995                .unwrap_or_default(),
996        };
997        // Repeatedly share a reference to the work until the channel is closed.
998        // The receiver will replace the data with the next work item, then wait for
999        // this thread to process it by calling into the C library and write the result.
1000        while channel.share(&mut work) {
1001            match work.command {
1002                AsyncCommand::None => {}
1003                AsyncCommand::Subscribe { qos } => {
1004                    let topic =
1005                        unsafe { core::ffi::CStr::from_bytes_with_nul_unchecked(&work.topic) };
1006                    work.result = client.subscribe_cstr(topic, qos);
1007                }
1008                AsyncCommand::Unsubscribe => {
1009                    let topic =
1010                        unsafe { core::ffi::CStr::from_bytes_with_nul_unchecked(&work.topic) };
1011                    work.result = client.unsubscribe_cstr(topic);
1012                }
1013                AsyncCommand::Publish { qos, retain } => {
1014                    let topic =
1015                        unsafe { core::ffi::CStr::from_bytes_with_nul_unchecked(&work.topic) };
1016                    work.result = client.publish_cstr(topic, qos, retain, &work.payload);
1017                }
1018                AsyncCommand::SetUri => {
1019                    let uri =
1020                        unsafe { core::ffi::CStr::from_bytes_with_nul_unchecked(&work.broker_uri) };
1021                    work.result = client.set_uri_cstr(uri);
1022                }
1023            }
1024        }
1025    }
1026}
1027
1028impl ErrorType for EspAsyncMqttClient {
1029    type Error = EspError;
1030}
1031
1032impl asynch::Client for EspAsyncMqttClient {
1033    async fn subscribe(&mut self, topic: &str, qos: QoS) -> Result<MessageId, Self::Error> {
1034        EspAsyncMqttClient::subscribe(self, topic, qos).await
1035    }
1036
1037    async fn unsubscribe(&mut self, topic: &str) -> Result<MessageId, Self::Error> {
1038        EspAsyncMqttClient::unsubscribe(self, topic).await
1039    }
1040}
1041
1042impl asynch::Publish for EspAsyncMqttClient {
1043    async fn publish(
1044        &mut self,
1045        topic: &str,
1046        qos: QoS,
1047        retain: bool,
1048        payload: &[u8],
1049    ) -> Result<MessageId, Self::Error> {
1050        EspAsyncMqttClient::publish(self, topic, qos, retain, payload).await
1051    }
1052}
1053
1054pub struct EspAsyncMqttConnection {
1055    receiver: Receiver<EspMqttEvent<'static>>,
1056    given: bool,
1057}
1058
1059impl EspAsyncMqttConnection {
1060    pub async fn next(&mut self) -> Result<&EspMqttEvent<'_>, EspError> {
1061        if self.given {
1062            self.receiver.done();
1063        }
1064
1065        if let Some(event) = self.receiver.get_shared_async().await {
1066            self.given = true;
1067
1068            Ok(event)
1069        } else {
1070            self.given = false;
1071
1072            Err(EspError::from_infallible::<ESP_ERR_INVALID_STATE>())
1073        }
1074    }
1075}
1076
1077impl ErrorType for EspAsyncMqttConnection {
1078    type Error = EspError;
1079}
1080
1081impl asynch::Connection for EspAsyncMqttConnection {
1082    type Event<'a> = &'a EspMqttEvent<'a>;
1083
1084    async fn next(&mut self) -> Result<Self::Event<'_>, Self::Error> {
1085        EspAsyncMqttConnection::next(self).await
1086    }
1087}
1088
1089static ERROR: EspError = EspError::from_infallible::<ESP_FAIL>();
1090
1091pub struct EspMqttEvent<'a>(&'a esp_mqtt_event_t);
1092
1093impl<'a> EspMqttEvent<'a> {
1094    const fn new(event: &'a esp_mqtt_event_t) -> Self {
1095        Self(event)
1096    }
1097
1098    #[allow(non_upper_case_globals, non_snake_case)]
1099    pub fn payload(&self) -> EventPayload<'_, EspError> {
1100        match self.0.event_id {
1101            esp_mqtt_event_id_t_MQTT_EVENT_ERROR => EventPayload::Error(&ERROR), // TODO
1102            esp_mqtt_event_id_t_MQTT_EVENT_BEFORE_CONNECT => EventPayload::BeforeConnect,
1103            esp_mqtt_event_id_t_MQTT_EVENT_CONNECTED => {
1104                EventPayload::Connected(self.0.session_present != 0)
1105            }
1106            esp_mqtt_event_id_t_MQTT_EVENT_DISCONNECTED => EventPayload::Disconnected,
1107            esp_mqtt_event_id_t_MQTT_EVENT_SUBSCRIBED => {
1108                EventPayload::Subscribed(self.0.msg_id as _)
1109            }
1110            esp_mqtt_event_id_t_MQTT_EVENT_UNSUBSCRIBED => {
1111                EventPayload::Unsubscribed(self.0.msg_id as _)
1112            }
1113            esp_mqtt_event_id_t_MQTT_EVENT_PUBLISHED => EventPayload::Published(self.0.msg_id as _),
1114            esp_mqtt_event_id_t_MQTT_EVENT_DATA => EventPayload::Received {
1115                id: self.0.msg_id as _,
1116                topic: {
1117                    let ptr = self.0.topic;
1118
1119                    if ptr.is_null() {
1120                        None
1121                    } else {
1122                        let len = self.0.topic_len;
1123
1124                        let topic = unsafe {
1125                            let slice = slice::from_raw_parts(ptr as _, len.try_into().unwrap());
1126                            core::str::from_utf8(slice).unwrap()
1127                        };
1128
1129                        Some(topic)
1130                    }
1131                },
1132                data: if self.0.data_len > 0 {
1133                    unsafe {
1134                        slice::from_raw_parts(
1135                            (self.0.data as *const u8).as_ref().unwrap(),
1136                            self.0.data_len as _,
1137                        )
1138                    }
1139                } else {
1140                    &[]
1141                },
1142                details: {
1143                    if self.0.data_len < self.0.total_data_len {
1144                        if self.0.current_data_offset == 0 {
1145                            Details::InitialChunk(InitialChunkData {
1146                                total_data_size: self.0.total_data_len as _,
1147                            })
1148                        } else {
1149                            Details::SubsequentChunk(SubsequentChunkData {
1150                                current_data_offset: self.0.current_data_offset as _,
1151                                total_data_size: self.0.total_data_len as _,
1152                            })
1153                        }
1154                    } else {
1155                        Details::Complete
1156                    }
1157                },
1158            },
1159            esp_mqtt_event_id_t_MQTT_EVENT_DELETED => EventPayload::Deleted(self.0.msg_id as _),
1160            other => panic!("Unknown message type: {other}"),
1161        }
1162    }
1163}
1164
1165/// SAFETY: EspMqttEvent contains no thread-specific data.
1166unsafe impl Send for EspMqttEvent<'_> {}
1167
1168/// SAFETY: EspMqttEvent is a read-only struct, so sharing it between threads is fine.
1169unsafe impl Sync for EspMqttEvent<'_> {}
1170
1171impl ErrorType for EspMqttEvent<'_> {
1172    type Error = EspError;
1173}
1174
1175impl Event for EspMqttEvent<'_> {
1176    fn payload(&self) -> EventPayload<'_, Self::Error> {
1177        EspMqttEvent::payload(self)
1178    }
1179}
esp_idf_svc/mqtt/client.rs

esp_idf_svc/mqtt/
client.rs
