mqtt_format/v3/
header.rs

1//
2//   This Source Code Form is subject to the terms of the Mozilla Public
3//   License, v. 2.0. If a copy of the MPL was not distributed with this
4//   file, You can obtain one at http://mozilla.org/MPL/2.0/.
5//
6
7use nom::{
8    bits,
9    bytes::complete::take_while_m_n,
10    error::{Error, ErrorKind, FromExternalError},
11    sequence::tuple,
12    IResult, Parser,
13};
14use nom_supreme::ParserExt;
15
16use super::{
17    errors::MPacketHeaderError,
18    qos::{mquality_of_service, MQualityOfService},
19};
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct MPacketHeader {
23    pub kind: MPacketKind,
24    pub remaining_length: u32,
25}
26
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum MPacketKind {
29    Connect,
30    Connack,
31    Publish {
32        dup: bool,
33        qos: MQualityOfService,
34        retain: bool,
35    },
36    Puback,
37    Pubrec,
38    Pubrel,
39    Pubcomp,
40    Subscribe,
41    Suback,
42    Unsubscribe,
43    Unsuback,
44    Pingreq,
45    Pingresp,
46    Disconnect,
47}
48
49impl MPacketKind {
50    pub fn to_byte(&self) -> u8 {
51        match self {
52            MPacketKind::Connect => 1 << 4,
53            MPacketKind::Connack => 2 << 4,
54            MPacketKind::Publish { dup, qos, retain } => {
55                let upper = 0b0011;
56                let dup = if *dup { 0b1000 } else { 0 };
57                let retain = u8::from(*retain);
58                let qos = match qos {
59                    MQualityOfService::AtMostOnce => 0b0000,
60                    MQualityOfService::AtLeastOnce => 0b0001,
61                    MQualityOfService::ExactlyOnce => 0b0010,
62                } << 1;
63
64                (upper << 4) | dup | retain | qos
65            }
66            MPacketKind::Puback => 4 << 4,
67            MPacketKind::Pubrec => 5 << 4,
68            MPacketKind::Pubrel => (6 << 4) | 2,
69            MPacketKind::Pubcomp => 7 << 4,
70            MPacketKind::Subscribe => (8 << 4) | 2,
71            MPacketKind::Suback => 9 << 4,
72            MPacketKind::Unsubscribe => (10 << 4) | 2,
73            MPacketKind::Unsuback => 11 << 4,
74            MPacketKind::Pingreq => 12 << 4,
75            MPacketKind::Pingresp => 13 << 4,
76            MPacketKind::Disconnect => 14 << 4,
77        }
78    }
79}
80
81fn mpacketkind(input: &[u8]) -> IResult<&[u8], MPacketKind> {
82    let (input, (upper, lower)): (_, (u8, u8)) =
83        bits::<_, _, Error<(&[u8], usize)>, _, _>(tuple((
84            nom::bits::complete::take(4usize),
85            nom::bits::complete::take(4usize),
86        )))(input)?;
87
88    let (input, kind) = match (upper, lower) {
89        (1, 0b0000) => (input, MPacketKind::Connect),
90        (2, 0b0000) => (input, MPacketKind::Connack),
91        (3, lower) => {
92            let dup = lower & 0b1000 != 0;
93            let retain = lower & 0b0001 != 0;
94            let qos = match mquality_of_service((lower & 0b0110) >> 1) {
95                Ok(qos) => qos,
96                Err(e) => {
97                    return Err(nom::Err::Error(Error::from_external_error(
98                        input,
99                        ErrorKind::MapRes,
100                        e,
101                    )));
102                }
103            };
104            (input, MPacketKind::Publish { qos, dup, retain })
105        }
106        (4, 0b0000) => (input, MPacketKind::Puback),
107        (5, 0b0000) => (input, MPacketKind::Pubrec),
108        (6, 0b0010) => (input, MPacketKind::Pubrel),
109        (7, 0b0000) => (input, MPacketKind::Pubcomp),
110        (8, 0b0010) => (input, MPacketKind::Subscribe),
111        (9, 0b0000) => (input, MPacketKind::Suback),
112        (10, 0b0010) => (input, MPacketKind::Unsubscribe),
113        (11, 0b0000) => (input, MPacketKind::Unsuback),
114        (12, 0b0000) => (input, MPacketKind::Pingreq),
115        (13, 0b0000) => (input, MPacketKind::Pingresp),
116        (14, 0b0000) => (input, MPacketKind::Disconnect),
117        (inv_type, _) => {
118            return Err(nom::Err::Error(Error::from_external_error(
119                input,
120                ErrorKind::MapRes,
121                MPacketHeaderError::InvalidPacketType(inv_type),
122            )))
123        }
124    };
125
126    Ok((input, kind))
127}
128
129pub fn decode_variable_length(bytes: &[u8]) -> u32 {
130    let mut output: u32 = 0;
131
132    for (exp, val) in bytes.iter().enumerate() {
133        output += (*val as u32 & 0b0111_1111) * 128u32.pow(exp as u32);
134    }
135
136    output
137}
138
139pub fn mfixedheader(input: &[u8]) -> IResult<&[u8], MPacketHeader> {
140    let (input, kind) = mpacketkind(input)?;
141    let (input, remaining_length) = nom::number::complete::u8
142        .and(take_while_m_n(0, 3, |b| b & 0b1000_0000 != 0))
143        .recognize()
144        .map(decode_variable_length)
145        .parse(input)?;
146
147    Ok((
148        input,
149        MPacketHeader {
150            kind,
151            remaining_length,
152        },
153    ))
154}
155
156#[cfg(test)]
157mod tests {
158    use crate::v3::{
159        header::{MPacketHeader, MPacketKind},
160        qos::MQualityOfService,
161    };
162
163    use super::{decode_variable_length, mfixedheader, mpacketkind};
164
165    #[test]
166    fn check_variable_length_decoding() {
167        let input = &[64];
168
169        let output = decode_variable_length(input);
170        assert_eq!(output, 64);
171
172        let input = &[193, 2];
173
174        let output = decode_variable_length(input);
175        assert_eq!(output, 321);
176    }
177
178    #[test]
179    fn check_header_publish_flags() {
180        let input = &[0b0011_1101, 0];
181
182        let (input, header) = mfixedheader(input).unwrap();
183
184        assert_eq!(input, &[]);
185
186        assert_eq!(
187            header,
188            MPacketHeader {
189                remaining_length: 0,
190                kind: MPacketKind::Publish {
191                    dup: true,
192                    qos: MQualityOfService::ExactlyOnce,
193                    retain: true
194                }
195            }
196        );
197    }
198
199    #[test]
200    fn check_invalid_header_publish_flags() {
201        let input = &[0b0011_1111, 0];
202
203        mfixedheader(input).unwrap_err();
204    }
205
206    #[test]
207    fn check_roundtrip_packet_kind() {
208        fn test(mp: MPacketKind) {
209            assert_eq!(mp, mpacketkind(&[mp.to_byte()]).unwrap().1);
210        }
211
212        test(MPacketKind::Connect);
213        test(MPacketKind::Connack);
214        test(MPacketKind::Publish {
215            dup: false,
216            qos: MQualityOfService::AtMostOnce,
217            retain: false,
218        });
219        test(MPacketKind::Publish {
220            dup: false,
221            qos: MQualityOfService::AtLeastOnce,
222            retain: false,
223        });
224        test(MPacketKind::Publish {
225            dup: false,
226            qos: MQualityOfService::ExactlyOnce,
227            retain: false,
228        });
229        test(MPacketKind::Publish {
230            dup: true,
231            qos: MQualityOfService::AtMostOnce,
232            retain: false,
233        });
234        test(MPacketKind::Publish {
235            dup: false,
236            qos: MQualityOfService::AtMostOnce,
237            retain: true,
238        });
239        test(MPacketKind::Puback);
240        test(MPacketKind::Pubrec);
241        test(MPacketKind::Pubrel);
242        test(MPacketKind::Pubcomp);
243        test(MPacketKind::Subscribe);
244        test(MPacketKind::Suback);
245        test(MPacketKind::Unsubscribe);
246        test(MPacketKind::Unsuback);
247        test(MPacketKind::Pingreq);
248        test(MPacketKind::Pingresp);
249        test(MPacketKind::Disconnect);
250    }
251}
mqtt_format/v3/header.rs

mqtt_format/v3/
header.rs
