mqtt_proto/v5/
publish.rs

1use core::convert::TryFrom;
2
3use alloc::string::String;
4use alloc::sync::Arc;
5use alloc::vec::Vec;
6
7use bytes::Bytes;
8use simdutf8::basic::from_utf8;
9
10use crate::{
11    from_read_exact_error, read_string, read_u16, read_u8, write_bytes, write_u16, write_u8,
12    AsyncRead, Encodable, Error, Pid, QoS, QosPid, SyncWrite, TopicName,
13};
14
15use super::{
16    decode_properties, encode_properties, encode_properties_len, ErrorV5, Header, PacketType,
17    UserProperty, VarByteInt,
18};
19
20/// Body type of PUBLISH packet.
21#[derive(Debug, Clone, PartialEq, Eq, Hash)]
22pub struct Publish {
23    pub dup: bool,
24    pub retain: bool,
25    pub qos_pid: QosPid,
26    pub topic_name: TopicName,
27    pub payload: Bytes,
28    pub properties: PublishProperties,
29}
30
31#[cfg(feature = "arbitrary")]
32impl<'a> arbitrary::Arbitrary<'a> for Publish {
33    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
34        Ok(Publish {
35            dup: u.arbitrary()?,
36            retain: u.arbitrary()?,
37            qos_pid: u.arbitrary()?,
38            topic_name: u.arbitrary()?,
39            properties: u.arbitrary()?,
40            payload: Bytes::from(Vec::<u8>::arbitrary(u)?),
41        })
42    }
43}
44
45impl Publish {
46    pub fn new(qos_pid: QosPid, topic_name: TopicName, payload: Bytes) -> Self {
47        Publish {
48            dup: false,
49            retain: false,
50            qos_pid,
51            topic_name,
52            payload,
53            properties: PublishProperties::default(),
54        }
55    }
56
57    pub async fn decode_async<T: AsyncRead + Unpin>(
58        reader: &mut T,
59        header: Header,
60    ) -> Result<Self, ErrorV5> {
61        let mut remaining_len = header.remaining_len as usize;
62        let topic_name = read_string(reader).await?;
63        remaining_len = remaining_len
64            .checked_sub(2 + topic_name.len())
65            .ok_or(Error::InvalidRemainingLength)?;
66        let qos_pid = match header.qos {
67            QoS::Level0 => QosPid::Level0,
68            QoS::Level1 => {
69                remaining_len = remaining_len
70                    .checked_sub(2)
71                    .ok_or(Error::InvalidRemainingLength)?;
72                QosPid::Level1(Pid::try_from(read_u16(reader).await?)?)
73            }
74            QoS::Level2 => {
75                remaining_len = remaining_len
76                    .checked_sub(2)
77                    .ok_or(Error::InvalidRemainingLength)?;
78                QosPid::Level2(Pid::try_from(read_u16(reader).await?)?)
79            }
80        };
81        let properties = PublishProperties::decode_async(reader, header.typ).await?;
82        remaining_len = remaining_len
83            .checked_sub(properties.encode_len())
84            .ok_or(Error::InvalidRemainingLength)?;
85        let payload = if remaining_len > 0 {
86            let mut data = alloc::vec![0u8; remaining_len];
87            reader
88                .read_exact(&mut data)
89                .await
90                .map_err(from_read_exact_error)?;
91            if properties.payload_is_utf8 == Some(true) && from_utf8(&data).is_err() {
92                return Err(ErrorV5::InvalidPayloadFormat);
93            }
94            data
95        } else {
96            Vec::new()
97        };
98        Ok(Publish {
99            dup: header.dup,
100            qos_pid,
101            retain: header.retain,
102            topic_name: TopicName::try_from(topic_name)?,
103            properties,
104            payload: Bytes::from(payload),
105        })
106    }
107}
108
109impl Encodable for Publish {
110    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
111        write_bytes(writer, self.topic_name.as_bytes())?;
112        match self.qos_pid {
113            QosPid::Level0 => {}
114            QosPid::Level1(pid) | QosPid::Level2(pid) => {
115                write_u16(writer, pid.value())?;
116            }
117        }
118        self.properties.encode(writer)?;
119        writer.write_all(self.payload.as_ref())?;
120        Ok(())
121    }
122
123    fn encode_len(&self) -> usize {
124        let mut len = 2 + self.topic_name.len();
125        match self.qos_pid {
126            QosPid::Level0 => {}
127            QosPid::Level1(_) | QosPid::Level2(_) => {
128                len += 2;
129            }
130        }
131        len += self.properties.encode_len();
132        len += self.payload.len();
133        len
134    }
135}
136
137/// Property list for PUBLISH packet.
138#[derive(Debug, Clone, PartialEq, Eq, Default, Hash)]
139pub struct PublishProperties {
140    pub payload_is_utf8: Option<bool>,
141    pub message_expiry_interval: Option<u32>,
142    pub topic_alias: Option<u16>,
143    pub response_topic: Option<TopicName>,
144    pub correlation_data: Option<Bytes>,
145    pub user_properties: Vec<UserProperty>,
146    // FIXME: this is a list of identifiers
147    pub subscription_id: Option<VarByteInt>,
148    pub content_type: Option<Arc<String>>,
149}
150
151#[cfg(feature = "arbitrary")]
152impl<'a> arbitrary::Arbitrary<'a> for PublishProperties {
153    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
154        Ok(PublishProperties {
155            payload_is_utf8: u.arbitrary()?,
156            message_expiry_interval: u.arbitrary()?,
157            topic_alias: u.arbitrary()?,
158            response_topic: u.arbitrary()?,
159            correlation_data: Option::<Vec<u8>>::arbitrary(u)?.map(Bytes::from),
160            user_properties: u.arbitrary()?,
161            subscription_id: u.arbitrary()?,
162            content_type: u.arbitrary()?,
163        })
164    }
165}
166
167impl PublishProperties {
168    pub async fn decode_async<T: AsyncRead + Unpin>(
169        reader: &mut T,
170        packet_type: PacketType,
171    ) -> Result<Self, ErrorV5> {
172        let mut properties = PublishProperties::default();
173        decode_properties!(
174            packet_type,
175            properties,
176            reader,
177            PayloadFormatIndicator,
178            MessageExpiryInterval,
179            TopicAlias,
180            ResponseTopic,
181            CorrelationData,
182            SubscriptionIdentifier,
183            ContentType,
184        );
185        Ok(properties)
186    }
187}
188
189impl Encodable for PublishProperties {
190    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
191        encode_properties!(
192            self,
193            writer,
194            PayloadFormatIndicator,
195            MessageExpiryInterval,
196            TopicAlias,
197            ResponseTopic,
198            CorrelationData,
199            SubscriptionIdentifier,
200            ContentType,
201        );
202        Ok(())
203    }
204    fn encode_len(&self) -> usize {
205        let mut len = 0;
206        encode_properties_len!(
207            self,
208            len,
209            PayloadFormatIndicator,
210            MessageExpiryInterval,
211            TopicAlias,
212            ResponseTopic,
213            CorrelationData,
214            SubscriptionIdentifier,
215            ContentType,
216        );
217        len
218    }
219}
220
221/// Body type for PUBACK packet.
222#[derive(Debug, Clone, PartialEq, Eq)]
223#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
224pub struct Puback {
225    pub pid: Pid,
226    pub reason_code: PubackReasonCode,
227    pub properties: PubackProperties,
228}
229
230impl Puback {
231    pub fn new(pid: Pid, reason_code: PubackReasonCode) -> Self {
232        Puback {
233            pid,
234            reason_code,
235            properties: PubackProperties::default(),
236        }
237    }
238
239    pub fn new_success(pid: Pid) -> Self {
240        Self::new(pid, PubackReasonCode::Success)
241    }
242
243    pub async fn decode_async<T: AsyncRead + Unpin>(
244        reader: &mut T,
245        header: Header,
246    ) -> Result<Self, ErrorV5> {
247        let pid = Pid::try_from(read_u16(reader).await?)?;
248        let (reason_code, properties) = if header.remaining_len == 2 {
249            (PubackReasonCode::Success, PubackProperties::default())
250        } else if header.remaining_len == 3 {
251            let reason_byte = read_u8(reader).await?;
252            let reason_code = PubackReasonCode::from_u8(reason_byte)
253                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
254            (reason_code, PubackProperties::default())
255        } else {
256            let reason_byte = read_u8(reader).await?;
257            let reason_code = PubackReasonCode::from_u8(reason_byte)
258                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
259            let properties = PubackProperties::decode_async(reader, header.typ).await?;
260            (reason_code, properties)
261        };
262        Ok(Puback {
263            pid,
264            reason_code,
265            properties,
266        })
267    }
268}
269
270impl Encodable for Puback {
271    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
272        write_u16(writer, self.pid.value())?;
273        if self.properties == PubackProperties::default() {
274            if self.reason_code != PubackReasonCode::Success {
275                write_u8(writer, self.reason_code as u8)?;
276            }
277        } else {
278            write_u8(writer, self.reason_code as u8)?;
279            self.properties.encode(writer)?;
280        }
281        Ok(())
282    }
283
284    fn encode_len(&self) -> usize {
285        if self.properties == PubackProperties::default() {
286            if self.reason_code == PubackReasonCode::Success {
287                2
288            } else {
289                3
290            }
291        } else {
292            3 + self.properties.encode_len()
293        }
294    }
295}
296
297/// Property list for PUBACK packet.
298#[derive(Debug, Clone, PartialEq, Eq, Default)]
299#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
300pub struct PubackProperties {
301    pub reason_string: Option<Arc<String>>,
302    pub user_properties: Vec<UserProperty>,
303}
304
305impl PubackProperties {
306    pub async fn decode_async<T: AsyncRead + Unpin>(
307        reader: &mut T,
308        packet_type: PacketType,
309    ) -> Result<Self, ErrorV5> {
310        let mut properties = PubackProperties::default();
311        decode_properties!(packet_type, properties, reader, ReasonString,);
312        Ok(properties)
313    }
314}
315
316impl Encodable for PubackProperties {
317    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
318        encode_properties!(self, writer, ReasonString,);
319        Ok(())
320    }
321    fn encode_len(&self) -> usize {
322        let mut len = 0;
323        encode_properties_len!(self, len, ReasonString,);
324        len
325    }
326}
327
328/// Reason code for PUBACK packet.
329///
330/// | Dec |  Hex | Reason Code name              | Description                                                                                                        |
331/// |-----|------|-------------------------------|--------------------------------------------------------------------------------------------------------------------|
332/// |   0 | 0x00 | Success                       | The message is accepted. Publication of the QoS 1 message proceeds.                                                |
333/// |  16 | 0x10 | No matching subscribers       | The message is accepted but there are no subscribers. This is sent only by the Server.                             |
334/// |     |      |                               | If the Server knows that there are no matching subscribers, it MAY use this Reason Code instead of 0x00 (Success). |
335/// | 128 | 0x80 | Unspecified error             | The receiver does not accept the publish but either does not want to reveal the reason,                            |
336/// |     |      |                               | or it does not match one of the other values.                                                                      |
337/// | 131 | 0x83 | Implementation specific error | The PUBLISH is valid but the receiver is not willing to accept it.                                                 |
338/// | 135 | 0x87 | Not authorized                | The PUBLISH is not authorized.                                                                                     |
339/// | 144 | 0x90 | Topic Name invalid            | The Topic Name is not malformed, but is not accepted by this Client or Server.                                     |
340/// | 145 | 0x91 | Packet identifier in use      | The Packet Identifier is already in use.                                                                           |
341/// |     |      |                               | This might indicate a mismatch in the Session State between the Client and Server.                                 |
342/// | 151 | 0x97 | Quota exceeded                | An implementation or administrative imposed limit has been exceeded.                                               |
343/// | 153 | 0x99 | Payload format invalid        | The payload format does not match the specified Payload Format Indicator.                                          |
344#[repr(u8)]
345#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
346#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
347pub enum PubackReasonCode {
348    Success = 0x00,
349    NoMatchingSubscribers = 0x10,
350    UnspecifiedError = 0x80,
351    ImplementationSpecificError = 0x83,
352    NotAuthorized = 0x87,
353    TopicNameInvalid = 0x90,
354    PacketIdentifierInUse = 0x91,
355    QuotaExceeded = 0x97,
356    PayloadFormatInvalid = 0x99,
357}
358
359impl PubackReasonCode {
360    pub fn from_u8(value: u8) -> Option<Self> {
361        let code = match value {
362            0x00 => Self::Success,
363            0x10 => Self::NoMatchingSubscribers,
364            0x80 => Self::UnspecifiedError,
365            0x83 => Self::ImplementationSpecificError,
366            0x87 => Self::NotAuthorized,
367            0x90 => Self::TopicNameInvalid,
368            0x91 => Self::PacketIdentifierInUse,
369            0x97 => Self::QuotaExceeded,
370            0x99 => Self::PayloadFormatInvalid,
371            _ => return None,
372        };
373        Some(code)
374    }
375}
376
377/// Body type for PUBREC packet.
378#[derive(Debug, Clone, PartialEq, Eq)]
379#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
380pub struct Pubrec {
381    pub pid: Pid,
382    pub reason_code: PubrecReasonCode,
383    pub properties: PubrecProperties,
384}
385
386impl Pubrec {
387    pub fn new(pid: Pid, reason_code: PubrecReasonCode) -> Self {
388        Pubrec {
389            pid,
390            reason_code,
391            properties: PubrecProperties::default(),
392        }
393    }
394
395    pub fn new_success(pid: Pid) -> Self {
396        Self::new(pid, PubrecReasonCode::Success)
397    }
398
399    pub async fn decode_async<T: AsyncRead + Unpin>(
400        reader: &mut T,
401        header: Header,
402    ) -> Result<Self, ErrorV5> {
403        let pid = Pid::try_from(read_u16(reader).await?)?;
404        let (reason_code, properties) = if header.remaining_len == 2 {
405            (PubrecReasonCode::Success, PubrecProperties::default())
406        } else if header.remaining_len == 3 {
407            let reason_byte = read_u8(reader).await?;
408            let reason_code = PubrecReasonCode::from_u8(reason_byte)
409                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
410            (reason_code, PubrecProperties::default())
411        } else {
412            let reason_byte = read_u8(reader).await?;
413            let reason_code = PubrecReasonCode::from_u8(reason_byte)
414                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
415            let properties = PubrecProperties::decode_async(reader, header.typ).await?;
416            (reason_code, properties)
417        };
418        Ok(Pubrec {
419            pid,
420            reason_code,
421            properties,
422        })
423    }
424}
425
426impl Encodable for Pubrec {
427    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
428        write_u16(writer, self.pid.value())?;
429        if self.properties == PubrecProperties::default() {
430            if self.reason_code != PubrecReasonCode::Success {
431                write_u8(writer, self.reason_code as u8)?;
432            }
433        } else {
434            write_u8(writer, self.reason_code as u8)?;
435            self.properties.encode(writer)?;
436        }
437        Ok(())
438    }
439
440    fn encode_len(&self) -> usize {
441        if self.properties == PubrecProperties::default() {
442            if self.reason_code == PubrecReasonCode::Success {
443                2
444            } else {
445                3
446            }
447        } else {
448            3 + self.properties.encode_len()
449        }
450    }
451}
452
453/// Property list for PUBREC packet.
454#[derive(Debug, Clone, PartialEq, Eq, Default)]
455#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
456pub struct PubrecProperties {
457    pub reason_string: Option<Arc<String>>,
458    pub user_properties: Vec<UserProperty>,
459}
460
461impl PubrecProperties {
462    pub async fn decode_async<T: AsyncRead + Unpin>(
463        reader: &mut T,
464        packet_type: PacketType,
465    ) -> Result<Self, ErrorV5> {
466        let mut properties = PubrecProperties::default();
467        decode_properties!(packet_type, properties, reader, ReasonString,);
468        Ok(properties)
469    }
470}
471
472impl Encodable for PubrecProperties {
473    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
474        encode_properties!(self, writer, ReasonString,);
475        Ok(())
476    }
477    fn encode_len(&self) -> usize {
478        let mut len = 0;
479        encode_properties_len!(self, len, ReasonString,);
480        len
481    }
482}
483
484/// Reason code for PUBREC packet.
485///
486/// | Dec |  Hex | Reason Code name              | Description                                                                                                        |
487/// |-----|------|-------------------------------|--------------------------------------------------------------------------------------------------------------------|
488/// |   0 | 0x00 | Success                       | The message is accepted. Publication of the QoS 2 message proceeds.                                                |
489/// |  16 | 0x10 | No matching subscribers       | The message is accepted but there are no subscribers. This is sent only by the Server.                             |
490/// |     |      |                               | If the Server knows that there are no matching subscribers, it MAY use this Reason Code instead of 0x00 (Success). |
491/// | 128 | 0x80 | Unspecified error             | The receiver does not accept the publish but either does not want to reveal the reason,                            |
492/// |     |      |                               | or it does not match one of the other values.                                                                      |
493/// | 131 | 0x83 | Implementation specific error | The PUBLISH is valid but the receiver is not willing to accept it.                                                 |
494/// | 135 | 0x87 | Not authorized                | The PUBLISH is not authorized.                                                                                     |
495/// | 144 | 0x90 | Topic Name invalid            | The Topic Name is not malformed, but is not accepted by this Client or Server.                                     |
496/// | 145 | 0x91 | Packet identifier in use      | The Packet Identifier is already in use.                                                                           |
497/// |     |      |                               | This might indicate a mismatch in the Session State between the Client and Server.                                 |
498/// | 151 | 0x97 | Quota exceeded                | An implementation or administrative imposed limit has been exceeded.                                               |
499/// | 153 | 0x99 | Payload format invalid        | The payload format does not match the specified Payload Format Indicator.                                          |
500#[repr(u8)]
501#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
502#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
503pub enum PubrecReasonCode {
504    Success = 0x00,
505    NoMatchingSubscribers = 0x10,
506    UnspecifiedError = 0x80,
507    ImplementationSpecificError = 0x83,
508    NotAuthorized = 0x87,
509    TopicNameInvalid = 0x90,
510    PacketIdentifierInUse = 0x91,
511    QuotaExceeded = 0x97,
512    PayloadFormatInvalid = 0x99,
513}
514
515impl PubrecReasonCode {
516    pub fn from_u8(value: u8) -> Option<Self> {
517        let code = match value {
518            0x00 => Self::Success,
519            0x10 => Self::NoMatchingSubscribers,
520            0x80 => Self::UnspecifiedError,
521            0x83 => Self::ImplementationSpecificError,
522            0x87 => Self::NotAuthorized,
523            0x90 => Self::TopicNameInvalid,
524            0x91 => Self::PacketIdentifierInUse,
525            0x97 => Self::QuotaExceeded,
526            0x99 => Self::PayloadFormatInvalid,
527            _ => return None,
528        };
529        Some(code)
530    }
531}
532
533/// Body type for PUBREL packet.
534#[derive(Debug, Clone, PartialEq, Eq)]
535#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
536pub struct Pubrel {
537    pub pid: Pid,
538    pub reason_code: PubrelReasonCode,
539    pub properties: PubrelProperties,
540}
541
542impl Pubrel {
543    pub fn new(pid: Pid, reason_code: PubrelReasonCode) -> Self {
544        Pubrel {
545            pid,
546            reason_code,
547            properties: PubrelProperties::default(),
548        }
549    }
550
551    pub fn new_success(pid: Pid) -> Self {
552        Self::new(pid, PubrelReasonCode::Success)
553    }
554
555    pub async fn decode_async<T: AsyncRead + Unpin>(
556        reader: &mut T,
557        header: Header,
558    ) -> Result<Self, ErrorV5> {
559        let pid = Pid::try_from(read_u16(reader).await?)?;
560        let (reason_code, properties) = if header.remaining_len == 2 {
561            (PubrelReasonCode::Success, PubrelProperties::default())
562        } else if header.remaining_len == 3 {
563            let reason_byte = read_u8(reader).await?;
564            let reason_code = PubrelReasonCode::from_u8(reason_byte)
565                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
566            (reason_code, PubrelProperties::default())
567        } else {
568            let reason_byte = read_u8(reader).await?;
569            let reason_code = PubrelReasonCode::from_u8(reason_byte)
570                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
571            let properties = PubrelProperties::decode_async(reader, header.typ).await?;
572            (reason_code, properties)
573        };
574        Ok(Pubrel {
575            pid,
576            reason_code,
577            properties,
578        })
579    }
580}
581
582impl Encodable for Pubrel {
583    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
584        write_u16(writer, self.pid.value())?;
585        if self.properties == PubrelProperties::default() {
586            if self.reason_code != PubrelReasonCode::Success {
587                write_u8(writer, self.reason_code as u8)?;
588            }
589        } else {
590            write_u8(writer, self.reason_code as u8)?;
591            self.properties.encode(writer)?;
592        }
593        Ok(())
594    }
595
596    fn encode_len(&self) -> usize {
597        if self.properties == PubrelProperties::default() {
598            if self.reason_code == PubrelReasonCode::Success {
599                2
600            } else {
601                3
602            }
603        } else {
604            3 + self.properties.encode_len()
605        }
606    }
607}
608
609/// Property list for PUBREL packet.
610#[derive(Debug, Clone, PartialEq, Eq, Default)]
611#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
612pub struct PubrelProperties {
613    pub reason_string: Option<Arc<String>>,
614    pub user_properties: Vec<UserProperty>,
615}
616
617impl PubrelProperties {
618    pub async fn decode_async<T: AsyncRead + Unpin>(
619        reader: &mut T,
620        packet_type: PacketType,
621    ) -> Result<Self, ErrorV5> {
622        let mut properties = PubrelProperties::default();
623        decode_properties!(packet_type, properties, reader, ReasonString,);
624        Ok(properties)
625    }
626}
627
628impl Encodable for PubrelProperties {
629    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
630        encode_properties!(self, writer, ReasonString,);
631        Ok(())
632    }
633    fn encode_len(&self) -> usize {
634        let mut len = 0;
635        encode_properties_len!(self, len, ReasonString,);
636        len
637    }
638}
639
640/// Reason code for PUBREL packet.
641///
642/// | Dec |  Hex | Reason Code name            | Description                                                                                 |
643/// |-----|------|-----------------------------|---------------------------------------------------------------------------------------------|
644/// |   0 | 0x00 | Success                     | Message released.                                                                           |
645/// | 146 | 0x92 | Packet Identifier not found | The Packet Identifier is not known. This is not an error during recovery,                   |
646/// |     |      |                             | but at other times indicates a mismatch between the Session State on the Client and Server. |
647#[repr(u8)]
648#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
649#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
650pub enum PubrelReasonCode {
651    Success = 0x00,
652    PacketIdentifierNotFound = 0x92,
653}
654
655impl PubrelReasonCode {
656    pub fn from_u8(value: u8) -> Option<Self> {
657        let code = match value {
658            0x00 => Self::Success,
659            0x92 => Self::PacketIdentifierNotFound,
660            _ => return None,
661        };
662        Some(code)
663    }
664}
665
666/// Body type for PUBCOMP packet.
667#[derive(Debug, Clone, PartialEq, Eq)]
668#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
669pub struct Pubcomp {
670    pub pid: Pid,
671    pub reason_code: PubcompReasonCode,
672    pub properties: PubcompProperties,
673}
674
675impl Pubcomp {
676    pub fn new(pid: Pid, reason_code: PubcompReasonCode) -> Self {
677        Pubcomp {
678            pid,
679            reason_code,
680            properties: PubcompProperties::default(),
681        }
682    }
683
684    pub fn new_success(pid: Pid) -> Self {
685        Self::new(pid, PubcompReasonCode::Success)
686    }
687
688    pub async fn decode_async<T: AsyncRead + Unpin>(
689        reader: &mut T,
690        header: Header,
691    ) -> Result<Self, ErrorV5> {
692        let pid = Pid::try_from(read_u16(reader).await?)?;
693        let (reason_code, properties) = if header.remaining_len == 2 {
694            (PubcompReasonCode::Success, PubcompProperties::default())
695        } else if header.remaining_len == 3 {
696            let reason_byte = read_u8(reader).await?;
697            let reason_code = PubcompReasonCode::from_u8(reason_byte)
698                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
699            (reason_code, PubcompProperties::default())
700        } else {
701            let reason_byte = read_u8(reader).await?;
702            let reason_code = PubcompReasonCode::from_u8(reason_byte)
703                .ok_or(ErrorV5::InvalidReasonCode(header.typ, reason_byte))?;
704            let properties = PubcompProperties::decode_async(reader, header.typ).await?;
705            (reason_code, properties)
706        };
707        Ok(Pubcomp {
708            pid,
709            reason_code,
710            properties,
711        })
712    }
713}
714
715impl Encodable for Pubcomp {
716    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
717        write_u16(writer, self.pid.value())?;
718        if self.properties == PubcompProperties::default() {
719            if self.reason_code != PubcompReasonCode::Success {
720                write_u8(writer, self.reason_code as u8)?;
721            }
722        } else {
723            write_u8(writer, self.reason_code as u8)?;
724            self.properties.encode(writer)?;
725        }
726        Ok(())
727    }
728
729    fn encode_len(&self) -> usize {
730        if self.properties == PubcompProperties::default() {
731            if self.reason_code == PubcompReasonCode::Success {
732                2
733            } else {
734                3
735            }
736        } else {
737            3 + self.properties.encode_len()
738        }
739    }
740}
741
742/// Property list for PUBCOMP packet.
743#[derive(Debug, Clone, PartialEq, Eq, Default)]
744#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
745pub struct PubcompProperties {
746    pub reason_string: Option<Arc<String>>,
747    pub user_properties: Vec<UserProperty>,
748}
749
750impl PubcompProperties {
751    pub async fn decode_async<T: AsyncRead + Unpin>(
752        reader: &mut T,
753        packet_type: PacketType,
754    ) -> Result<Self, ErrorV5> {
755        let mut properties = PubcompProperties::default();
756        decode_properties!(packet_type, properties, reader, ReasonString,);
757        Ok(properties)
758    }
759}
760
761impl Encodable for PubcompProperties {
762    fn encode<W: SyncWrite>(&self, writer: &mut W) -> Result<(), Error> {
763        encode_properties!(self, writer, ReasonString,);
764        Ok(())
765    }
766    fn encode_len(&self) -> usize {
767        let mut len = 0;
768        encode_properties_len!(self, len, ReasonString,);
769        len
770    }
771}
772
773/// Reason code for PUBCOMP packet.
774///
775/// | Dec |  Hex | Reason Code name            | Description                                                                                 |
776/// |-----|------|-----------------------------|---------------------------------------------------------------------------------------------|
777/// |   0 | 0x00 | Success                     | Packet Identifier released. Publication of QoS 2 message is complete.                       |
778/// | 146 | 0x92 | Packet Identifier not found | The Packet Identifier is not known. This is not an error during recovery,                   |
779/// |     |      |                             | but at other times indicates a mismatch between the Session State on the Client and Server. |
780#[repr(u8)]
781#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
782#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
783pub enum PubcompReasonCode {
784    Success = 0x00,
785    PacketIdentifierNotFound = 0x92,
786}
787
788impl PubcompReasonCode {
789    pub fn from_u8(value: u8) -> Option<Self> {
790        let code = match value {
791            0x00 => Self::Success,
792            0x92 => Self::PacketIdentifierNotFound,
793            _ => return None,
794        };
795        Some(code)
796    }
797}
mqtt_proto/v5/publish.rs

mqtt_proto/v5/
publish.rs
