flowsdk 0.4.2

Safety-first, realistic, behavior-predictable messaging SDK for MQTT and more.
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

// SPDX-License-Identifier: MPL-2.0

use super::control_packet::MqttPacket;
use crate::mqtt_serde::base_data::{BinaryData, TwoByteInteger, Utf8String, VariableByteInteger};
use std::error::Error;
use std::fmt;
use std::io::Error as IoError;

pub type ParserResult = Result<ParseOk, ParseError>;

// First byte of Fixed header
pub const FIXED_HDR_LEN: usize = 1;

#[derive(Debug)]
pub enum ParseError {
    More(usize, String), // not enough data for processing, hint for how many more bytes are needed
    IoError(IoError),
    ParseError(String),
    IncompleteProperty,
    // following two are for UTF-8 parsing errors
    Utf8Error(std::str::Utf8Error),
    FromUtf8Error(std::string::FromUtf8Error),
    StringTooLong,
    BufferTooShort,
    BufferEmpty,
    InvalidLength,
    InvalidPropertyId,
    InvalidPacketType,
    UnSuppProtoVsn,
    InternalError(String),
    InvalidVariableByteInteger,
}

impl fmt::Display for ParseError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            ParseError::More(hint, msg) => write!(f, "More data needed ({} bytes): {}", hint, msg),
            ParseError::IoError(e) => write!(f, "IO Error: {}", e),
            ParseError::ParseError(msg) => write!(f, "Parse Error: {}", msg),
            ParseError::IncompleteProperty => write!(f, "Incomplete Property"),
            ParseError::Utf8Error(e) => write!(f, "UTF-8 Error: {}", e),
            ParseError::FromUtf8Error(e) => write!(f, "From UTF-8 Error: {}", e),
            ParseError::StringTooLong => write!(f, "String Too Long"),
            ParseError::BufferTooShort => write!(f, "Buffer Too Short"),
            ParseError::BufferEmpty => write!(f, "Buffer is Empty"),
            ParseError::InvalidLength => write!(f, "Invalid Length"),
            ParseError::InvalidPropertyId => write!(f, "Invalid Property ID"),
            ParseError::InvalidPacketType => write!(f, "Invalid Packet Type"),
            ParseError::UnSuppProtoVsn => write!(f, "Unsupported Protocol Version"),
            ParseError::InternalError(msg) => write!(f, "Internal Error: {}", msg),
            ParseError::InvalidVariableByteInteger => write!(f, "Invalid Variable Byte Integer"),
        }
    }
}

impl Error for ParseError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            ParseError::IoError(e) => Some(e),
            ParseError::Utf8Error(e) => Some(e),
            ParseError::FromUtf8Error(e) => Some(e),
            _ => None,
        }
    }
}

#[derive(Debug, PartialEq)]
#[allow(clippy::large_enum_variant)]
pub enum ParseOk {
    Continue(usize, usize),    // (hint, consumed)
    Packet(MqttPacket, usize), // (packet, consumed)
    TopicName(String, usize),  // (topic_name, consumed)
}

pub fn packet_type(buffer: &[u8]) -> Result<u8, ParseError> {
    if buffer.is_empty() {
        return Err(ParseError::BufferEmpty);
    }
    Ok(buffer[0] >> 4)
}

pub fn parse_remaining_length(buffer: &[u8]) -> Result<(usize, usize), ParseError> {
    vbi(buffer)
}

// 1.5.4 UTF-8 Encoded String
pub fn parse_utf8_string(buffer: &[u8]) -> Result<(String, usize), ParseError> {
    Utf8String::decode(buffer)
}

pub fn parse_topic_name(buffer: &[u8]) -> Result<ParseOk, ParseError> {
    let (res, consumed) = parse_utf8_string(buffer)?;
    Ok(ParseOk::TopicName(res.to_string(), consumed))
}

pub fn parse_packet_id(buffer: &[u8]) -> Result<(u16, usize), ParseError> {
    TwoByteInteger::decode(buffer)
}

pub fn parse_binary_data(buffer: &[u8]) -> Result<(Vec<u8>, usize), ParseError> {
    BinaryData::decode(buffer)
}

pub fn parse_vbi(buffer: &[u8]) -> Result<(usize, usize), ParseError> {
    vbi(buffer)
}

// Helper functions for backward compatibility with MQTT v3 modules
pub fn read_string(buffer: &[u8]) -> Result<(String, usize), ParseError> {
    parse_utf8_string(buffer)
}

pub fn write_string(s: &str) -> Result<Vec<u8>, ParseError> {
    Ok(Utf8String::encode(s))
}

pub fn read_u8(buffer: &[u8]) -> Result<u8, ParseError> {
    if buffer.is_empty() {
        return Err(ParseError::BufferTooShort);
    }
    Ok(buffer[0])
}

/*                   */
/* Internal helpers  */
/*                   */

// Variable Byte Integer
fn vbi(buffer: &[u8]) -> Result<(usize, usize), ParseError> {
    let (value, consumed) = VariableByteInteger::decode(buffer)?;
    Ok((value, consumed))
}

pub mod stream;