rust_mqtt/packet/v5/
subscription_packet.rs1use heapless::Vec;
26
27use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
28use crate::packet::v5::mqtt_packet::Packet;
29use crate::packet::v5::publish_packet::QualityOfService;
30use crate::utils::buffer_reader::BuffReader;
31use crate::utils::buffer_writer::BuffWriter;
32use crate::utils::types::{BufferError, TopicFilter};
33
34use super::packet_type::PacketType;
35use super::property::Property;
36
37pub struct SubscriptionPacket<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> {
38 pub fixed_header: u8,
39 pub remain_len: u32,
40 pub packet_identifier: u16,
41 pub property_len: u32,
42 pub properties: Vec<Property<'a>, MAX_PROPERTIES>,
43 pub topic_filter_len: u16,
44 pub topic_filters: Vec<TopicFilter<'a>, MAX_FILTERS>,
45}
46
47impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize>
48 SubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES>
49{
50 pub fn add_new_filter(&mut self, topic_name: &'a str, qos: QualityOfService) {
51 let len = topic_name.len();
52 let mut new_filter = TopicFilter::new();
53 new_filter.filter.string = topic_name;
54 new_filter.filter.len = len as u16;
55 new_filter.sub_options |= <QualityOfService as Into<u8>>::into(qos) >> 1;
56 self.topic_filters.push(new_filter);
57 self.topic_filter_len += 1;
58 }
59}
60
61impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a>
62 for SubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES>
63{
64 fn new() -> Self {
65 Self {
66 fixed_header: PacketType::Subscribe.into(),
67 remain_len: 0,
68 packet_identifier: 1,
69 property_len: 0,
70 properties: Vec::<Property<'a>, MAX_PROPERTIES>::new(),
71 topic_filter_len: 0,
72 topic_filters: Vec::<TopicFilter<'a>, MAX_FILTERS>::new(),
73 }
74 }
75
76 fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result<usize, BufferError> {
77 let mut buff_writer = BuffWriter::new(buffer, buffer_len);
78
79 let mut rm_ln = self.property_len;
80 let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?;
81 let property_len_len = VariableByteIntegerEncoder::len(property_len_enc);
82
83 let mut lt = 0;
84 let mut filters_len = 0;
85 loop {
86 filters_len = filters_len + self.topic_filters.get(lt).unwrap().filter.len + 3;
87 lt += 1;
88 if lt == self.topic_filter_len as usize {
89 break;
90 }
91 }
92 rm_ln = rm_ln + property_len_len as u32 + 2 + filters_len as u32;
93
94 buff_writer.write_u8(self.fixed_header)?;
95 buff_writer.write_variable_byte_int(rm_ln)?;
96 buff_writer.write_u16(self.packet_identifier)?;
97 buff_writer.write_variable_byte_int(self.property_len)?;
98 buff_writer.write_properties::<MAX_PROPERTIES>(&self.properties)?;
99 buff_writer.write_topic_filters_ref(
100 true,
101 self.topic_filter_len as usize,
102 &self.topic_filters,
103 )?;
104 Ok(buff_writer.position)
105 }
106
107 fn decode(&mut self, _buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> {
108 error!("Subscribe packet does not support decode funtion on client!");
109 Err(BufferError::WrongPacketToDecode)
110 }
111 fn set_property_len(&mut self, value: u32) {
112 self.property_len = value;
113 }
114
115 fn get_property_len(&mut self) -> u32 {
116 self.property_len
117 }
118
119 fn push_to_properties(&mut self, property: Property<'a>) {
120 self.properties.push(property);
121 }
122
123 fn property_allowed(&mut self, property: &Property<'a>) -> bool {
124 property.subscribe_property()
125 }
126
127 fn set_fixed_header(&mut self, header: u8) {
128 self.fixed_header = header;
129 }
130
131 fn set_remaining_len(&mut self, remaining_len: u32) {
132 self.remain_len = remaining_len;
133 }
134}