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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
use std::time::Duration;
use crate::util::constants::DEFAULT_RECEIVE_MAXIMUM;
use crate::{
packets::{ConnectProperties, LastWill},
util::constants::MAXIMUM_PACKET_SIZE,
};
#[derive(Debug, Clone, Copy, thiserror::Error)]
pub enum ConnectOptionsError {
#[error("Maximum packet size is exceeded. Maximum is {MAXIMUM_PACKET_SIZE}, user provided: {0}")]
MaximumPacketSizeExceeded(u32),
}
/// Options for the connection to the MQTT broker
#[derive(Debug, Clone)]
pub struct ConnectOptions {
/// client identifier
client_id: Box<str>,
/// keep alive time to send pingreq to broker when the connection is idle
pub(crate) keep_alive_interval: Duration,
/// clean or persistent session indicator
pub(crate) clean_start: bool,
/// username and password
username: Option<Box<str>>,
password: Option<Box<str>>,
// MQTT v5 Connect Properties:
session_expiry_interval: Option<u32>,
/// The maximum number of packets that will be inflight from the broker to this client.
receive_maximum: Option<u16>,
/// The maximum number of packets that can be inflight from this client to the broker.
send_maximum: Option<u16>,
maximum_packet_size: Option<u32>,
topic_alias_maximum: Option<u16>,
request_response_information: Option<u8>,
request_problem_information: Option<u8>,
user_properties: Vec<(Box<str>, Box<str>)>,
authentication_method: Option<Box<str>>,
authentication_data: Option<Vec<u8>>,
/// Last will that will be issued on unexpected disconnect
last_will: Option<LastWill>,
}
impl Default for ConnectOptions {
fn default() -> Self {
Self {
keep_alive_interval: Duration::from_secs(60),
clean_start: true,
client_id: Box::from("ChangeClientId_MQRSTT"),
username: None,
password: None,
session_expiry_interval: None,
receive_maximum: None,
send_maximum: None,
maximum_packet_size: None,
topic_alias_maximum: None,
request_response_information: None,
request_problem_information: None,
user_properties: Vec::new(),
authentication_method: None,
authentication_data: None,
last_will: None,
}
}
}
impl ConnectOptions {
/// Create a new [`ConnectOptions`]
///
/// Be aware:
/// This client does not restrict the client identifier in any way. However, the MQTT v5.0 specification does.
/// It is thus recommended to use a client id that is compatible with the MQTT v5.0 specification.
/// - 1 to 23 bytes UTF-8 bytes.
/// - Contains [a-zA-Z0-9] characters only.
///
/// Some brokers accept longer client ids with different characters
pub fn new<S: AsRef<str>>(client_id: S) -> Self {
Self {
keep_alive_interval: Duration::from_secs(60),
clean_start: true,
client_id: client_id.as_ref().into(),
username: None,
password: None,
session_expiry_interval: None,
receive_maximum: None,
send_maximum: None,
maximum_packet_size: None,
topic_alias_maximum: None,
request_response_information: None,
request_problem_information: None,
user_properties: vec![],
authentication_method: None,
authentication_data: None,
last_will: None,
}
}
pub(crate) fn create_connect_from_options(&self) -> crate::packets::Packet {
let connect_properties = ConnectProperties {
session_expiry_interval: self.session_expiry_interval,
receive_maximum: self.receive_maximum,
maximum_packet_size: self.maximum_packet_size,
topic_alias_maximum: self.topic_alias_maximum,
request_response_information: self.request_response_information,
request_problem_information: self.request_problem_information,
user_properties: self.user_properties.clone(),
authentication_method: self.authentication_method.clone(),
authentication_data: self.authentication_data.clone(),
};
let connect = crate::packets::Connect {
client_id: self.client_id.clone(),
clean_start: self.clean_start,
keep_alive: self.keep_alive_interval.as_secs() as u16,
username: self.username.clone(),
password: self.password.clone(),
connect_properties,
protocol_version: crate::packets::ProtocolVersion::V5,
last_will: self.last_will.clone(),
};
crate::packets::Packet::Connect(connect)
}
/// The Client Identifier (ClientID) identifies the Client to the Server. Each Client connecting to the Server has a unique ClientID.
/// The ClientID MUST be used by Clients and by Servers to identify state that they hold relating to this MQTT Session between the Client and the Server [MQTT-3.1.3-2].
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059>
///
/// Non unique client ids often result in connect, disconnect loops due to auto reconnects and forced disconnects
pub fn get_client_id(&self) -> &str {
&self.client_id
}
/// The Client Identifier (ClientID) identifies the Client to the Server. Each Client connecting to the Server has a unique ClientID.
/// The ClientID MUST be used by Clients and by Servers to identify state that they hold relating to this MQTT Session between the Client and the Server [MQTT-3.1.3-2].
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059>
///
/// Non unique client ids often result in connect, disconnect loops due to auto reconnects and forced disconnects
pub fn set_client_id<S: AsRef<str>>(&mut self, client_id: S) -> &mut Self {
self.client_id = client_id.as_ref().into();
self
}
/// This specifies whether the Connection starts a new Session or is a continuation of an existing Session.
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901039>
pub fn get_clean_start(&self) -> bool {
self.clean_start
}
/// This specifies whether the Connection starts a new Session or is a continuation of an existing Session.
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901039>
pub fn set_clean_start(&mut self, clean_start: bool) -> &mut Self {
self.clean_start = clean_start;
self
}
pub fn get_username(&self) -> Option<&str> {
self.username.as_ref().map(Box::<str>::as_ref)
}
pub fn set_username<S: AsRef<str>>(&mut self, username: S) -> &mut Self {
self.username = Some(username.as_ref().into());
self
}
pub fn get_password(&self) -> Option<&str> {
self.password.as_ref().map(Box::<str>::as_ref)
}
pub fn set_password<S: AsRef<str>>(&mut self, password: S) -> &mut Self {
self.password = Some(password.as_ref().into());
self
}
/// Get the Session Expiry Interval in seconds.
/// If the Session Expiry Interval is absent the value 0 is used. If it is set to 0, or is absent, the Session ends when the Network Connection is closed.
/// If the Session Expiry Interval is 0xFFFFFFFF (UINT_MAX), the Session does not expire.
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901211>
pub fn get_session_expiry_interval(&self) -> Option<u32> {
self.session_expiry_interval
}
/// Set the Session Expiry Interval in seconds.
/// If the Session Expiry Interval is absent the value 0 is used. If it is set to 0, or is absent, the Session ends when the Network Connection is closed.
/// If the Session Expiry Interval is 0xFFFFFFFF (UINT_MAX), the Session does not expire.
/// More info here: <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901211>
pub fn set_session_expiry_interval(&mut self, session_expiry_interval: u32) -> &mut Self {
self.session_expiry_interval = Some(session_expiry_interval);
self
}
/// Get the current keep alive interval for the MQTT protocol
/// This time is used in Ping Pong when natural traffic is absent
/// The granularity is in seconds!
pub fn get_keep_alive_interval(&self) -> Duration {
self.keep_alive_interval
}
/// Set the keep alive interval for the MQTT protocol
/// This time is used in Ping Pong when natural traffic is absent
/// The granularity is in seconds!
pub fn set_keep_alive_interval(&mut self, keep_alive_interval: Duration) -> &mut Self {
self.keep_alive_interval = Duration::from_secs(keep_alive_interval.as_secs());
self
}
pub fn set_last_will(&mut self, last_will: LastWill) -> &mut Self {
self.last_will = Some(last_will);
self
}
pub fn get_last_will(&self) -> Option<&LastWill> {
self.last_will.as_ref()
}
pub fn set_receive_maximum(&mut self, receive_maximum: u16) -> &mut Self {
self.receive_maximum = Some(receive_maximum);
self
}
pub fn receive_maximum(&self) -> u16 {
self.receive_maximum.unwrap_or(DEFAULT_RECEIVE_MAXIMUM)
}
pub fn set_maximum_packet_size(&mut self, maximum_packet_size: u32) -> Result<&mut Self, ConnectOptionsError> {
if maximum_packet_size > MAXIMUM_PACKET_SIZE {
Err(ConnectOptionsError::MaximumPacketSizeExceeded(maximum_packet_size))
} else {
self.maximum_packet_size = Some(maximum_packet_size);
Ok(self)
}
}
pub fn maximum_packet_size(&self) -> usize {
self.maximum_packet_size.unwrap_or(MAXIMUM_PACKET_SIZE) as usize
}
pub fn set_send_maximum(&mut self, send_maximum: u16) -> &mut Self {
self.send_maximum = Some(send_maximum);
self
}
pub fn send_maximum(&self) -> u16 {
self.send_maximum.unwrap_or(DEFAULT_RECEIVE_MAXIMUM)
}
}