rumqttc/mqttbytes/v4/
suback.rs

1use super::*;
2use bytes::{Buf, BufMut, Bytes, BytesMut};
3use std::convert::{TryFrom, TryInto};
4
5/// Acknowledgement to subscribe
6#[derive(Debug, Clone, PartialEq, Eq)]
7pub struct SubAck {
8    pub pkid: u16,
9    pub return_codes: Vec<SubscribeReasonCode>,
10}
11
12impl SubAck {
13    pub fn new(pkid: u16, return_codes: Vec<SubscribeReasonCode>) -> SubAck {
14        SubAck { pkid, return_codes }
15    }
16
17    fn len(&self) -> usize {
18        2 + self.return_codes.len()
19    }
20
21    pub fn size(&self) -> usize {
22        let len = self.len();
23        let remaining_len_size = len_len(len);
24        1 + remaining_len_size + len
25    }
26
27    pub fn read(fixed_header: FixedHeader, mut bytes: Bytes) -> Result<Self, Error> {
28        let variable_header_index = fixed_header.fixed_header_len;
29        bytes.advance(variable_header_index);
30        let pkid = read_u16(&mut bytes)?;
31
32        if !bytes.has_remaining() {
33            return Err(Error::MalformedPacket);
34        }
35
36        let mut return_codes = Vec::new();
37        while bytes.has_remaining() {
38            let return_code = read_u8(&mut bytes)?;
39            return_codes.push(return_code.try_into()?);
40        }
41
42        let suback = SubAck { pkid, return_codes };
43        Ok(suback)
44    }
45
46    pub fn write(&self, buffer: &mut BytesMut) -> Result<usize, Error> {
47        buffer.put_u8(0x90);
48        let remaining_len = self.len();
49        let remaining_len_bytes = write_remaining_length(buffer, remaining_len)?;
50
51        buffer.put_u16(self.pkid);
52        let p: Vec<u8> = self
53            .return_codes
54            .iter()
55            .map(|&code| match code {
56                SubscribeReasonCode::Success(qos) => qos as u8,
57                SubscribeReasonCode::Failure => 0x80,
58            })
59            .collect();
60        buffer.extend_from_slice(&p);
61        Ok(1 + remaining_len_bytes + remaining_len)
62    }
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum SubscribeReasonCode {
67    Success(QoS),
68    Failure,
69}
70
71impl TryFrom<u8> for SubscribeReasonCode {
72    type Error = super::Error;
73
74    fn try_from(value: u8) -> Result<Self, Self::Error> {
75        let v = match value {
76            0 => SubscribeReasonCode::Success(QoS::AtMostOnce),
77            1 => SubscribeReasonCode::Success(QoS::AtLeastOnce),
78            2 => SubscribeReasonCode::Success(QoS::ExactlyOnce),
79            128 => SubscribeReasonCode::Failure,
80            v => return Err(super::Error::InvalidSubscribeReasonCode(v)),
81        };
82
83        Ok(v)
84    }
85}
86
87#[cfg(test)]
88mod test {
89    use super::*;
90    use bytes::BytesMut;
91    use pretty_assertions::assert_eq;
92
93    #[test]
94    fn suback_parsing_works() {
95        let stream = vec![
96            0x90, 4, // packet type, flags and remaining len
97            0x00, 0x0F, // variable header. pkid = 15
98            0x01, 0x80, // payload. return codes [success qos1, failure]
99            0xDE, 0xAD, 0xBE, 0xEF, // extra packets in the stream
100        ];
101
102        let mut stream = BytesMut::from(&stream[..]);
103        let fixed_header = parse_fixed_header(stream.iter()).unwrap();
104        let ack_bytes = stream.split_to(fixed_header.frame_length()).freeze();
105        let packet = SubAck::read(fixed_header, ack_bytes).unwrap();
106
107        assert_eq!(
108            packet,
109            SubAck {
110                pkid: 15,
111                return_codes: vec![
112                    SubscribeReasonCode::Success(QoS::AtLeastOnce),
113                    SubscribeReasonCode::Failure,
114                ],
115            }
116        );
117    }
118}
rumqttc/mqttbytes/v4/suback.rs

rumqttc/mqttbytes/v4/
suback.rs
