Dreaded_X/rust-mqtt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Complete encode

Browse Source
This commit is contained in:
Ondrej Babec 2022-02-21 15:08:56 +01:00
parent 6f0d2d5de4
commit 963040026a
No known key found for this signature in database
GPG Key ID: 13E577E3769B2079
19 changed files with 162 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/main.rs
View File

@ -4,7 +4,9 @@ use rust_mqtt::packet::property::*;*/
use std::fs::File;
use std::io::Read;*/
use rust_mqtt::packet::connect_packet::ConnectPacket;
use rust_mqtt::packet::mqtt_packet::Packet;
use rust_mqtt::packet::publish_packet::PublishPacket;
use rust_mqtt::packet::subscription_packet::SubscriptionPacket;
fn main() {
@ -14,9 +16,20 @@ fn main() {
.init();
let mut pckt: SubscriptionPacket<1> = SubscriptionPacket::new();
let mut res = vec![0; 14];
pckt.encode(&mut res);
println!("{:02X?}", res);
let mut res = vec![0; 140];
let lnsub = pckt.encode(&mut res);
println!("{:02X?}", &res[0..lnsub]);
let mut res2 = vec![0; 260];
let mut x = b"hello world";
let mut pblsh: PublishPacket = PublishPacket::new(x);
let lnpblsh = pblsh.encode(&mut res2);
println!("{:02X?}", &res2[0..lnpblsh]);
log::info!("xxx");
let mut res3 = vec![0; 260];
let mut cntrl = ConnectPacket::<3, 0>::clean();
let lncntrl = cntrl.encode(&mut res3);
println!("{:02X?}", &res3[0..lncntrl]);
log::info!("xxx");
/*let fl = File::open("/Users/obabec/development/school/rust-mqtt/mqtt_control_example.bin");

3
src/packet/auth_packet.rs
View File

@ -50,7 +50,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI
return AuthPacket { fixed_header: PacketType::Auth.into(), remain_len: 0, auth_reason: 0, property_len: 0, properties: Vec::<Property<'a>, MAX_PROPERTIES>::new() }
}*/
fn encode(&mut self, buffer: & mut [u8]) {
fn encode(&mut self, buffer: & mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -64,6 +64,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI
buff_writer.write_u8(self.auth_reason);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/connack_packet.rs
View File

@ -34,7 +34,7 @@ impl<'a> ConnackPacket<'a> {
}
impl<'a> Packet<'a> for ConnackPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
buff_writer.write_u8(self.fixed_header);
let mut property_len_enc = VariableByteIntegerEncoder::encode(self.property_len).unwrap();
@ -46,6 +46,7 @@ impl<'a> Packet<'a> for ConnackPacket<'a> {
buff_writer.write_u8(self.connect_reason_code);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {
self.decode_connack_packet(buff_reader);

134
src/packet/control_packet.rs → src/packet/connect_packet.rs
View File

@ -1,26 +1,21 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::utils::buffer_reader::BinaryData;
use crate::utils::buffer_reader::BuffReader;
use crate::utils::buffer_reader::EncodedString;
use crate::utils::buffer_reader::ParseError;
use crate::utils::buffer_writer::BuffWriter;
use super::packet_type::PacketType;
use super::property::Property;
pub const MAX_PROPERTIES: usize = 18;
pub const MAX_WILL_PROPERTIES: usize = 7;
pub struct ControlPacket<'a> {
pub struct ConnectPacket<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> {
// 7 - 4 mqtt control packet type, 3-0 flagy
pub fixed_header: u8,
// 1 - 4 B lenght of variable header + len of payload
pub remain_len: u32,
// variable header
//optional prida se pouze u packetu ve kterych ma co delat
pub packet_identifier: u16,
pub protocol_name_len: u16,
pub protocol_name: u32,
pub protocol_version: u8,
@ -43,30 +38,31 @@ pub struct ControlPacket<'a> {
pub password: BinaryData<'a>,
}
impl<'a> ControlPacket<'a> {
pub fn clean(
properties: Vec<Property<'a>, MAX_PROPERTIES>,
will_properties: Vec<Property<'a>, MAX_WILL_PROPERTIES>,
) -> Self {
Self {
fixed_header: 0x00,
impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> ConnectPacket<'a, MAX_PROPERTIES, MAX_WILL_PROPERTIES> {
pub fn clean() -> Self {
let mut x = Self {
fixed_header: PacketType::Connect.into(),
remain_len: 0,
packet_identifier: 0,
protocol_name_len: 0,
protocol_name: 0,
protocol_name_len: 4,
protocol_name: 0x4d515454,
protocol_version: 5,
connect_flags: 0,
keep_alive: 0,
property_len: 0,
properties,
connect_flags: 0x02,
keep_alive: 60,
property_len: 3,
properties: Vec::<Property<'a>, MAX_PROPERTIES>::new(),
client_id: EncodedString::new(),
will_property_len: 0,
will_properties,
will_properties: Vec::<Property<'a>, MAX_WILL_PROPERTIES>::new(),
will_topic: EncodedString::new(),
will_payload: BinaryData::new(),
username: EncodedString::new(),
password: BinaryData::new(),
}
};
let y = Property::ReceiveMaximum(20);
x.properties.push(y);
x.client_id.len = 0;
return x;
}
pub fn get_reason_code(&self) {
@ -99,67 +95,63 @@ impl<'a> ControlPacket<'a> {
}
self.fixed_header = cur_type | flags;
}
}
impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<'a> for ConnectPacket<'a, MAX_PROPERTIES, MAX_WILL_PROPERTIES> {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
pub fn decode_will_properties(&mut self, buff_reader: &mut BuffReader<'a>) {
//todo: need to check if we are parsing only will properties
let will_property_len = buff_reader.read_variable_byte_int().unwrap();
let mut x: u32 = 0;
let mut prop: Result<Property, ParseError>;
loop {
let mut res: Property;
prop = Property::decode(buff_reader);
if let Ok(res) = prop {
log::info!("Will property parsed: {:?}", res);
x = x + res.len() as u32 + 1;
self.will_properties.push(res);
} else {
// error handlo
log::error!("Problem during property decoding");
let mut rm_ln = self.property_len;
let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len).unwrap();
let property_len_len = VariableByteIntegerEncoder::len(property_len_enc);
// 12 = protocol_name_len + protocol_name + protocol_version + connect_flags + keep_alive + client_id_len
rm_ln = rm_ln + property_len_len as u32 + 12;
if self.connect_flags & 0x04 == 1 {
let wil_prop_len_enc = VariableByteIntegerEncoder::encode(self.will_property_len).unwrap();
let wil_prop_len_len = VariableByteIntegerEncoder::len(wil_prop_len_enc);
rm_ln = rm_ln + wil_prop_len_len as u32 + self.will_property_len as u32 + self.will_topic.len as u32 + self.will_payload.len as u32;
}
if x == will_property_len {
break;
}
}
if self.connect_flags & 0x80 == 1 {
rm_ln = rm_ln + self.username.len as u32;
}
pub fn decode_payload(&mut self, buff_reader: &mut BuffReader<'a>) {
self.client_id = buff_reader.read_string().unwrap();
if self.connect_flags & (1 << 2) == 1 {
self.decode_will_properties(buff_reader);
self.will_topic = buff_reader.read_string().unwrap();
self.will_payload = buff_reader.read_binary().unwrap();
if self.connect_flags & 0x40 == 1 {
rm_ln = rm_ln + self.password.len as u32;
}
if self.connect_flags & (1 << 7) == 1 {
self.username = buff_reader.read_string().unwrap();
}
if self.connect_flags & (1 << 6) == 1 {
self.password = buff_reader.read_binary().unwrap();
}
buff_writer.write_u8(self.fixed_header);
buff_writer.write_variable_byte_int(rm_ln);
buff_writer.write_u16(self.protocol_name_len);
buff_writer.write_u32(self.protocol_name);
buff_writer.write_u8(self.protocol_version);
buff_writer.write_u8(self.connect_flags);
buff_writer.write_u16(self.keep_alive);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
buff_writer.write_string_ref(&self.client_id);
if self.connect_flags & 0x04 == 1 {
buff_writer.write_variable_byte_int(self.will_property_len);
buff_writer.encode_properties(& self.will_properties);
buff_writer.write_string_ref(& self.will_topic);
buff_writer.write_binary_ref(& self.will_payload);
}
pub fn decode_control_packet(&mut self, buff_reader: &mut BuffReader<'a>) {
if self.decode_fixed_header(buff_reader) != (PacketType::Connect).into() {
log::error!("Packet you are trying to decode is not CONNECT packet!");
}
self.packet_identifier = 0;
self.protocol_name_len = buff_reader.read_u16().unwrap();
self.protocol_name = buff_reader.read_u32().unwrap();
self.protocol_version = buff_reader.read_u8().unwrap();
self.connect_flags = buff_reader.read_u8().unwrap();
self.keep_alive = buff_reader.read_u16().unwrap();
self.decode_properties(buff_reader);
self.decode_payload(buff_reader);
}
if self.connect_flags & 0x80 == 1 {
buff_writer.write_string_ref(& self.username);
}
impl<'a> Packet<'a> for ControlPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {}
if self.connect_flags & 0x40 == 1 {
buff_writer.write_binary_ref(& self.password);
}
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {
log::error!("Decode function is not available for control packet!")
//self.decode_control_packet(buff_reader);
}
fn set_property_len(&mut self, value: u32) {

3
src/packet/disconnect_packet.rs
View File

@ -33,7 +33,7 @@ impl<'a, const MAX_PROPERTIES: usize> DisconnectPacket<'a, MAX_PROPERTIES> {
}
impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for DisconnectPacket<'a, MAX_PROPERTIES> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
buff_writer.write_u8(self.fixed_header);
let mut property_len_enc = VariableByteIntegerEncoder::encode(self.property_len).unwrap();
@ -44,6 +44,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for DisconnectPacket<'a, MAX_PR
buff_writer.write_u8(self.disconnect_reason);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

1
src/packet/mod.rs
View File

@ -19,3 +19,4 @@ pub mod pingreq_packet;
pub mod pingresp_packet;
pub mod suback_packet;
pub mod unsuback_packet;
pub mod connect_packet;

2
src/packet/mqtt_packet.rs
View File

@ -7,7 +7,7 @@ use super::property::Property;
pub trait Packet<'a> {
//fn new() -> dyn Packet<'a> where Self: Sized;
fn encode(&mut self, buffer: &mut [u8]);
fn encode(&mut self, buffer: &mut [u8]) -> usize;
fn decode(&mut self, buff_reader: &mut BuffReader<'a>);
// properties

3
src/packet/pingreq_packet.rs
View File

@ -17,10 +17,11 @@ pub struct PingreqPacket {
impl PingreqPacket {}
impl<'a> Packet<'a> for PingreqPacket {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
buff_writer.write_u8(self.fixed_header);
buff_writer.write_variable_byte_int(0 as u32);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/pingresp_packet.rs
View File

@ -22,10 +22,11 @@ impl<'a> PingrespPacket {
}
impl<'a> Packet<'a> for PingrespPacket {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
buff_writer.write_u8(self.fixed_header);
buff_writer.write_variable_byte_int(0 as u32);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/puback_packet.rs
View File

@ -38,7 +38,7 @@ impl<'a> PubackPacket<'a> {
}
impl<'a> Packet<'a> for PubackPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -52,6 +52,7 @@ impl<'a> Packet<'a> for PubackPacket<'a> {
buff_writer.write_u8(self.reason_code);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/pubcomp_packet.rs
View File

@ -39,7 +39,7 @@ impl<'a> PubcompPacket<'a> {
}
impl<'a> Packet<'a> for PubcompPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -53,6 +53,7 @@ impl<'a> Packet<'a> for PubcompPacket<'a> {
buff_writer.write_u8(self.reason_code);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

28
src/packet/publish_packet.rs
View File

@ -29,35 +29,57 @@ pub struct PublishPacket<'a> {
}
impl<'a> PublishPacket<'a> {
pub fn new(message: &'a [u8]) -> Self {
let mut x = Self { fixed_header: PacketType::Publish.into(), remain_len: 0, topic_name: EncodedString::new(), packet_identifier: 0,
property_len: 0, properties: Vec::<Property<'a>, MAX_PROPERTIES>::new(), message };
x.topic_name.string = "test/topic";
x.topic_name.len = 10;
return x;
}
pub fn decode_publish_packet(&mut self, buff_reader: &mut BuffReader<'a>) {
if self.decode_fixed_header(buff_reader) != (PacketType::Publish).into() {
log::error!("Packet you are trying to decode is not PUBLISH packet!");
return;
}
self.topic_name = buff_reader.read_string().unwrap();
let qos = self.fixed_header & 0x03;
if qos != 0 {
// je potreba dekodovat jenom pro QoS 1 / 2
self.packet_identifier = buff_reader.read_u16().unwrap();
}
self.decode_properties(buff_reader);
self.message = buff_reader.read_message();
}
}
impl<'a> Packet<'a> for PublishPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len).unwrap();
let property_len_len = VariableByteIntegerEncoder::len(property_len_enc);
let mut msg_len = self.message.len() as u32;
rm_ln = rm_ln + property_len_len as u32 + 2 + msg_len + self.topic_name.len as u32 + 2;
rm_ln = rm_ln + property_len_len as u32 + msg_len + self.topic_name.len as u32 + 2;
buff_writer.write_u8(self.fixed_header);
let qos = self.fixed_header & 0x03;
if qos != 0 {
rm_ln + 2;
}
buff_writer.write_variable_byte_int(rm_ln);
buff_writer.write_string_ref(& self.topic_name);
if qos != 0 {
buff_writer.write_u16(self.packet_identifier);
}
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
buff_writer.insert_ref(msg_len as usize, self.message)
buff_writer.insert_ref(msg_len as usize, self.message);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/pubrec_packet.rs
View File

@ -39,7 +39,7 @@ impl<'a> PubrecPacket<'a> {
}
impl<'a> Packet<'a> for PubrecPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -53,6 +53,7 @@ impl<'a> Packet<'a> for PubrecPacket<'a> {
buff_writer.write_u8(self.reason_code);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/pubrel_packet.rs
View File

@ -39,7 +39,7 @@ impl<'a> PubrelPacket<'a> {
}
impl<'a> Packet<'a> for PubrelPacket<'a> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -53,6 +53,7 @@ impl<'a> Packet<'a> for PubrelPacket<'a> {
buff_writer.write_u8(self.reason_code);
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/suback_packet.rs
View File

@ -51,8 +51,9 @@ impl<'a, const MAX_REASONS: usize> SubackPacket<'a, MAX_REASONS> {
}
impl<'a, const MAX_REASONS: usize> Packet<'a> for SubackPacket<'a, MAX_REASONS> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
log::error!("SUBACK packet does not support encoding!");
return 0;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/subscription_packet.rs
View File

@ -44,7 +44,7 @@ impl<'a, const MAX_FILTERS: usize> SubscriptionPacket<'a, MAX_FILTERS> {
}
impl<'a, const MAX_FILTERS: usize> Packet<'a> for SubscriptionPacket<'a, MAX_FILTERS> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -68,6 +68,7 @@ impl<'a, const MAX_FILTERS: usize> Packet<'a> for SubscriptionPacket<'a, MAX_FIL
buff_writer.write_variable_byte_int(self.property_len);
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
buff_writer.encode_topic_filters_ref(false, self.topic_filter_len as usize, & self.topic_filters);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/unsuback_packet.rs
View File

@ -49,8 +49,9 @@ impl<'a, const MAX_REASONS: usize> UnsubackPacket<'a, MAX_REASONS> {
}
impl<'a, const MAX_REASONS: usize> Packet<'a> for UnsubackPacket<'a, MAX_REASONS> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
log::error!("UNSUBACK packet does not support encoding!");
return 0;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

3
src/packet/unsubscription_packet.rs
View File

@ -38,7 +38,7 @@ impl<'a, const MAX_FILTERS: usize> UnsubscriptionPacket<'a, MAX_FILTERS> {
}
impl<'a, const MAX_FILTERS: usize> Packet<'a> for UnsubscriptionPacket<'a, MAX_FILTERS> {
fn encode(&mut self, buffer: &mut [u8]) {
fn encode(&mut self, buffer: &mut [u8]) -> usize {
let mut buff_writer = BuffWriter::new(buffer);
let mut rm_ln = self.property_len;
@ -53,6 +53,7 @@ impl<'a, const MAX_FILTERS: usize> Packet<'a> for UnsubscriptionPacket<'a, MAX_F
buff_writer.encode_properties::<MAX_PROPERTIES>(&self.properties);
buff_writer.write_u16(self.topic_filter_len);
buff_writer.encode_topic_filters_ref(false, self.topic_filter_len as usize, & self.topic_filters);
return buff_writer.position;
}
fn decode(&mut self, buff_reader: &mut BuffReader<'a>) {

9
src/utils/buffer_writer.rs
View File

@ -13,6 +13,7 @@ impl<'a> BuffWriter<'a> {
pub fn insert<const LEN: usize>(& mut self, array: [u8; LEN]) {
let mut x: usize = 0;
if LEN != 0 {
loop {
self.buffer[self.position] = array[x];
self.increment_position(1);
@ -22,9 +23,11 @@ impl<'a> BuffWriter<'a> {
}
}
}
}
pub fn insert_ref(& mut self, len: usize, array: &[u8]) {
let mut x: usize = 0;
if len != 0 {
loop {
self.buffer[self.position] = array[x];
self.increment_position(1);
@ -34,6 +37,7 @@ impl<'a> BuffWriter<'a> {
}
}
}
}
pub fn new(buffer: &'a mut [u8]) -> Self {
return BuffWriter {
@ -107,12 +111,13 @@ impl<'a> BuffWriter<'a> {
pub fn encode_properties<const LEN: usize>(& mut self, properties: & Vec<Property<'a>, LEN>) {
let mut i = 0;
if properties.len() != 0 {
let len = properties.len();
if len != 0 {
loop {
let prop: &Property = properties.get(i).unwrap();
self.encode_property(prop);
i = i + 1;
if i == LEN {
if i == len {
break;
}
}