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
|
use futures::future::TryFutureExt;
use librumqttd::{async_locallink, Config};
use mqtt_client::{Client, Message, MqttClient, MqttClientError, QoS, Topic, TopicFilter};
use rumqttc::StateError;
use tokio::time::Duration;
#[derive(Debug)]
enum TestJoinError {
TestMqttClientError(MqttClientError),
ElapseTime,
}
#[tokio::test]
// This checks the mqtt packets are within the limit or not
async fn packet_size_within_limit() -> Result<(), anyhow::Error> {
// Start the local broker
let mqtt_server_handle = tokio::spawn(async {
start_broker_local("../../configuration/rumqttd/rumqttd_5883.conf").await
});
// Start the subscriber
let subscriber = tokio::spawn(async move { subscribe_until_3_messages_received().await });
// Start the publisher and publish 3 messages
let publisher = tokio::spawn(async move { publish_3_messages().await });
let _ = publisher.await?;
let res = subscriber.await?;
mqtt_server_handle.abort();
match res {
Err(e) => {
return Err(e);
}
_ => {
return Ok(());
}
}
}
#[tokio::test]
// This checks the mqtt packet size that exceeds the limit
async fn packet_size_exceeds_limit() -> Result<(), anyhow::Error> {
// Start the broker
let mqtt_server_handle = tokio::spawn(async {
start_broker_local("../../configuration/rumqttd/rumqttd_5884.conf").await
});
// Start the publisher and publish a message
let publish = tokio::spawn(async { publish_big_message_wait_for_error().await });
// if error is received then test is ok, else test should fail
let res = publish.await?;
mqtt_server_handle.abort();
match res {
Err(e) => {
return Err(e);
}
_ => {
return Ok(());
}
}
}
async fn subscribe_errors(pub_client: &Client) -> Result<(), MqttClientError> {
let mut errors = pub_client.subscribe_errors();
// return particular error else return Ok
while let Some(error) = errors.next().await {
match *error {
MqttClientError::ConnectionError(rumqttc::ConnectionError::MqttState(
StateError::Deserialization(rumqttc::Error::PayloadTooLong),
)) => {
return Err(mqtt_client::MqttClientError::ConnectionError(
rumqttc::ConnectionError::Mqtt4Bytes(rumqttc::Error::PayloadTooLong),
));
}
_ => {
return Ok(());
}
}
}
Ok(())
}
async fn start_broker_local(cfile: &str) -> anyhow::Result<()> {
let config: Config = confy::load_path(cfile)?;
let (mut router, _console, servers, _builder) = async_locallink::construct_broker(config);
let router = tokio::task::spawn_blocking(move || -> anyhow::Result<()> { Ok(router.start()?) });
servers.await;
let _ = router.await;
Ok(())
}
async fn subscribe_until_3_messages_received() -> Result<(), anyhow::Error> {
let sub_filter = TopicFilter::new("test/hello")?;
let client =
Client::connect("subscribe", &mqtt_client::Config::default().with_port(5883)).await?;
let mut messages = client.subscribe(sub_filter).await?;
let mut cnt: i32 = 0;
while let Some(_message) = messages.next().await {
if cnt >= 3 {
break;
} else {
cnt += 1;
}
}
assert!(cnt >= 3);
client.disconnect().await?;
Ok(())
}
async fn publish_3_messages() -> Result<(), anyhow::Error> {
// create a 128MB message
let buffer = create_packet(134217728);
let topic = Topic::new("test/hello")?;
let client = Client::connect(
"publish_big_data",
&mqtt_client::Config::default().with_port(5883),
)
.await?;
let message = Message::new(&topic, buffer.clone()).qos(QoS::AtMostOnce);
let mut cnt: i32 = 0;
loop {
let () = client.publish(message.clone()).await?;
tokio::time::sleep(Duration::from_secs(1)).await;
if cnt >= 3 {
break;
} else {
cnt += 1;
}
}
client.disconnect().await?;
Ok(())
}
async fn publish_big_message_wait_for_error() -> Result<(), anyhow::Error> {
// create a 260MB message
let buffer = create_packet(272629760);
let topic = Topic::new("test/hello")?;
let publish_client = Client::connect(
"publish_big_data",
&mqtt_client::Config::default().with_port(5884),
)
.await?;
let message = Message::new(&topic, buffer.clone()).qos(QoS::ExactlyOnce);
let publish_handle = publish_client.publish(message);
// wait for error else timeout
let timeout = tokio::time::timeout(
std::time::Duration::from_secs(2),
subscribe_errors(&publish_client).map_err(|e| TestJoinError::TestMqttClientError(e)),
)
.map_err(|_e| TestJoinError::ElapseTime);
// wait until one of the future returns error
let res = tokio::try_join!(
timeout,
publish_handle.map_err(|e| TestJoinError::TestMqttClientError(e))
);
match res {
Ok((first, _second)) => match first {
Err(TestJoinError::TestMqttClientError(_)) => {
return Ok(());
}
_ => {
anyhow::bail!("Did not catch error correctly");
}
},
_ => {
anyhow::bail!("test failed");
}
}
}
fn create_packet(size: usize) -> String {
let data: String = "Some data!".into();
let loops = size / data.len();
let mut buffer = String::with_capacity(size);
for _ in 0..loops {
buffer.push_str("Some data!");
}
buffer
}
|
