Feature/add helper mqtt methods

Cargo.toml

@@ -14,3 +14,16 @@ categories = ["embedded"]
 
 [workspace]
 members = ["mqtt-format", "mqtt-tester"]
+
+[dependencies]
+futures = "0.3.30"
+mqtt-format = { version = "0.5.0", path = "mqtt-format", features = [
+    "yoke",
+    "mqttv5",
+] }
+stable_deref_trait = "1.2.0"
+thiserror = "1.0.58"
+tokio = { version = "1.36.0", features = ["macros", "full"] }
+tokio-util = { version = "0.7.10", features = ["codec"] }
+winnow = "0.6.5"
+yoke = "0.7.3"

mqtt-format/src/v5/fixed_header.rs

@@ -104,6 +104,10 @@ impl MFixedHeader {
         Ok(MFixedHeader { packet_type })
     }
 
+    pub fn binary_size() -> u32 {
+        1
+    }
+
     pub fn write<W: WriteMqttPacket>(&self, buffer: &mut W) -> WResult<W> {
         #[allow(clippy::identity_op)]
         let byte = match self.packet_type {

mqtt-format/src/v5/integers.rs

@@ -56,17 +56,24 @@ pub fn write_u32<W: WriteMqttPacket>(buffer: &mut W, u: u32) -> WResult<W> {
 /// The maximal value is smaller than a u32, so that type is used
 ///
 #[doc = crate::v5::util::md_speclink!("_Toc3901011")]
-pub fn parse_variable_u32(input: &mut &Bytes) -> MResult<u32> {
-    trace("mqtt_variable_u32", |input: &mut &Bytes| {
+pub fn parse_variable_u32<'a, I>(input: &mut I) -> MResult<u32>
+where
+    I: winnow::stream::StreamIsPartial + winnow::stream::Stream<Token = u8>,
+    <I as winnow::stream::Stream>::Slice: IntoIterator<Item = &'a u8>,
+{
+    trace("mqtt_variable_u32", |input: &mut I| {
         let var_bytes = (
-            take_while(0..=3, |b| b & 0b1000_0000 != 0),
-            winnow::binary::u8.verify(|b: &u8| b & 0b1000_0000 == 0),
+            take_while(0..=3, |b| b & 0b1000_0000 != 0)
+                .context(winnow::error::StrContext::Label("front bytes")),
+            winnow::binary::u8
+                .verify(|b: &u8| b & 0b1000_0000 == 0)
+                .context(winnow::error::StrContext::Label("last byte")),
         );
-        let bytes: &[u8] = var_bytes.recognize().parse_next(input)?;
+        let bytes = var_bytes.recognize().parse_next(input)?;
 
         let mut output: u32 = 0;
 
-        for (exp, val) in bytes.iter().enumerate() {
+        for (exp, val) in bytes.into_iter().enumerate() {
             output += (*val as u32 & 0b0111_1111) * 128u32.pow(exp as u32);
         }
 
@@ -126,7 +133,12 @@ pub fn write_variable_u32<W: WriteMqttPacket>(buffer: &mut W, u: u32) -> WResult
 
 #[cfg(test)]
 mod tests {
+    use core::num::NonZeroUsize;
+
+    use pretty_assertions::assert_eq;
+    use winnow::error::ErrMode;
     use winnow::Bytes;
+    use winnow::Partial;
 
     use crate::v5::integers::parse_u16;
     use crate::v5::integers::parse_u32;
@@ -135,6 +147,30 @@ mod tests {
     use crate::v5::test::TestWriter;
     use crate::v5::write::WriteMqttPacket;
 
+    #[test]
+    fn check_incomplete_parsing() {
+        let input = [0xFF];
+
+        assert_eq!(
+            parse_variable_u32(&mut Partial::new(Bytes::new(&input))).unwrap_err(),
+            ErrMode::Incomplete(winnow::error::Needed::Size(NonZeroUsize::new(1).unwrap()))
+        );
+
+        let input = [0xFF, 0xFF, 0xFF];
+
+        assert_eq!(
+            parse_variable_u32(&mut Partial::new(Bytes::new(&input))).unwrap_err(),
+            ErrMode::Incomplete(winnow::error::Needed::Size(NonZeroUsize::new(1).unwrap()))
+        );
+
+        let input = [0xFF, 0xFF, 0xFF, 0xFF];
+
+        assert!(matches!(
+            parse_variable_u32(&mut Partial::new(Bytes::new(&input))).unwrap_err(),
+            ErrMode::Backtrack(_)
+        ));
+    }
+
     #[test]
     fn check_integer_parsing() {
         let input = 15u16.to_be_bytes();

mqtt-format/src/v5/packets/mod.rs

@@ -48,7 +48,7 @@ pub mod unsuback;
 pub mod unsubscribe;
 
 #[cfg_attr(feature = "yoke", derive(yoke::Yokeable))]
-#[derive(Clone, Debug)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum MqttPacket<'i> {
     Auth(MAuth<'i>),
     Connack(MConnack<'i>),
@@ -107,6 +107,30 @@ impl<'i> MqttPacket<'i> {
         Self::parse(&mut Bytes::new(input))
     }
 
+    pub fn binary_size(&self) -> u32 {
+        let header = MFixedHeader::binary_size();
+
+        let packet_size = match self {
+            MqttPacket::Auth(packet) => packet.binary_size(),
+            MqttPacket::Connack(packet) => packet.binary_size(),
+            MqttPacket::Connect(packet) => packet.binary_size(),
+            MqttPacket::Disconnect(packet) => packet.binary_size(),
+            MqttPacket::Pingreq(packet) => packet.binary_size(),
+            MqttPacket::Pingresp(packet) => packet.binary_size(),
+            MqttPacket::Puback(packet) => packet.binary_size(),
+            MqttPacket::Pubcomp(packet) => packet.binary_size(),
+            MqttPacket::Publish(packet) => packet.binary_size(),
+            MqttPacket::Pubrec(packet) => packet.binary_size(),
+            MqttPacket::Pubrel(packet) => packet.binary_size(),
+            MqttPacket::Suback(packet) => packet.binary_size(),
+            MqttPacket::Subscribe(packet) => packet.binary_size(),
+            MqttPacket::Unsuback(packet) => packet.binary_size(),
+            MqttPacket::Unsubscribe(packet) => packet.binary_size(),
+        };
+
+        header + crate::v5::integers::variable_u32_binary_size(packet_size) + packet_size
+    }
+
     pub fn write<W: WriteMqttPacket>(&self, buffer: &mut W) -> WResult<W> {
         match self {
             MqttPacket::Auth(p) => {