mqrstt 0.4.2

Pure rust MQTTv5 client implementation Smol and Tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

mod properties;

pub use properties::SubAckProperties;

mod reason_code;
pub use reason_code::SubAckReasonCode;

use super::{
    VariableInteger, WireLength,
    error::SerializeError,
    mqtt_trait::{MqttAsyncRead, MqttRead, MqttWrite, PacketAsyncRead, PacketRead, PacketWrite},
};
use bytes::BufMut;
use tokio::io::AsyncReadExt;

/// SubAck packet is sent by the server in response to a [`crate::packets::Subscribe`] packet.
///
/// A SUBACK packet is sent by the Server to the Client to confirm receipt and processing of a SUBSCRIBE packet.
/// A SUBACK packet contains a list of Reason Codes, that specify the maximum QoS level that was granted or the error which was found for each Subscription that was requested by the SUBSCRIBE.
#[derive(Debug, Default, PartialEq, Eq, Clone)]
pub struct SubAck {
    pub packet_identifier: u16,
    pub properties: SubAckProperties,
    pub reason_codes: Vec<SubAckReasonCode>,
}

impl PacketRead for SubAck {
    fn read(_: u8, remaining_length: usize, mut buf: bytes::Bytes) -> Result<Self, super::error::DeserializeError> {
        let packet_identifier = u16::read(&mut buf)?;
        let properties = SubAckProperties::read(&mut buf)?;

        let mut reason_codes = vec![];

        let mut read = 2 + properties.wire_len().variable_integer_len() + properties.wire_len();
        loop {
            if read >= remaining_length {
                break;
            }

            let reason_code = SubAckReasonCode::read(&mut buf)?;
            reason_codes.push(reason_code);
            read += 1;
        }

        Ok(Self {
            packet_identifier,
            properties,
            reason_codes,
        })
    }
}

impl<S> PacketAsyncRead<S> for SubAck
where
    S: tokio::io::AsyncRead + Unpin,
{
    async fn async_read(_: u8, remaining_length: usize, stream: &mut S) -> Result<(Self, usize), crate::packets::error::ReadError> {
        let mut total_read_bytes = 0;
        let packet_identifier = stream.read_u16().await?;
        let (properties, proproperties_read_bytes) = SubAckProperties::async_read(stream).await?;
        total_read_bytes += 2 + proproperties_read_bytes;
        let mut reason_codes = vec![];
        loop {
            if remaining_length == total_read_bytes {
                break;
            }

            let (reason_code, reason_code_read_bytes) = SubAckReasonCode::async_read(stream).await?;
            total_read_bytes += reason_code_read_bytes;
            reason_codes.push(reason_code);
        }

        Ok((
            Self {
                packet_identifier,
                properties,
                reason_codes,
            },
            total_read_bytes,
        ))
    }
}

impl PacketWrite for SubAck {
    fn write(&self, buf: &mut bytes::BytesMut) -> Result<(), SerializeError> {
        buf.put_u16(self.packet_identifier);

        self.properties.write(buf)?;
        for reason_code in &self.reason_codes {
            reason_code.write(buf)?;
        }

        Ok(())
    }
}

impl<S> crate::packets::mqtt_trait::PacketAsyncWrite<S> for SubAck
where
    S: tokio::io::AsyncWrite + Unpin,
{
    async fn async_write(&self, stream: &mut S) -> Result<usize, crate::packets::error::WriteError> {
        use crate::packets::mqtt_trait::MqttAsyncWrite;
        use tokio::io::AsyncWriteExt;
        let mut total_written_bytes = 2;
        stream.write_u16(self.packet_identifier).await?;

        total_written_bytes += self.properties.async_write(stream).await?;

        for reason_code in &self.reason_codes {
            reason_code.async_write(stream).await?;
        }
        total_written_bytes += self.reason_codes.len();

        Ok(total_written_bytes)
    }
}

impl WireLength for SubAck {
    fn wire_len(&self) -> usize {
        2 + self.properties.wire_len().variable_integer_len() + self.properties.wire_len() + self.reason_codes.len()
    }
}

#[cfg(test)]
mod test {
    use bytes::BytesMut;

    use super::SubAck;
    use crate::packets::mqtt_trait::{PacketRead, PacketWrite};

    #[test]
    fn read_write_suback() {
        let buf = vec![
            0x00, 0x0F, // variable header. pkid = 15
            0x00, // Property length 0
            0x01, // Payload reason code codes Granted QoS 1,
            0x80, // Payload Unspecified error
        ];

        let data = BytesMut::from(&buf[..]);
        let sub_ack = SubAck::read(0, 5, data.clone().into()).unwrap();

        let mut result = BytesMut::new();
        sub_ack.write(&mut result).unwrap();

        assert_eq!(data.to_vec(), result.to_vec());
    }
}