Skip to content

Commit e63bcab

Browse files
author
Devdutt Shenoi
committed
refactor: roll out Network::connect
1 parent a1282cd commit e63bcab

File tree

4 files changed

+20
-43
lines changed

4 files changed

+20
-43
lines changed

rumqttc/src/eventloop.rs

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ impl EventLoop {
134134

135135
requests_in_channel.retain(|request| {
136136
match request {
137-
Request::PubAck(_) => false, // Wait for publish retransmission, else the broker could be confused by an unexpected ack
137+
Request::PubAck(_) => false, // Wait for publish retransmission, else the broker could be confused by an unexpected ack
138138
_ => true,
139139
}
140140
});
@@ -486,18 +486,15 @@ async fn mqtt_connect(
486486
options: &MqttOptions,
487487
network: &mut Network,
488488
) -> Result<ConnAck, ConnectionError> {
489-
let keep_alive = options.keep_alive().as_secs() as u16;
490-
let clean_session = options.clean_session();
491-
let last_will = options.last_will();
492-
493489
let mut connect = Connect::new(options.client_id());
494-
connect.keep_alive = keep_alive;
495-
connect.clean_session = clean_session;
496-
connect.last_will = last_will;
490+
connect.keep_alive = options.keep_alive().as_secs() as u16;
491+
connect.clean_session = options.clean_session();
492+
connect.last_will = options.last_will();
497493
connect.login = options.credentials();
498494

499495
// send mqtt connect packet
500-
network.connect(connect).await?;
496+
network.write(Packet::Connect(connect)).await?;
497+
network.flush().await?;
501498

502499
// validate connack
503500
match network.read().await? {

rumqttc/src/framed.rs

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -84,12 +84,6 @@ impl Network {
8484
.map_err(StateError::Deserialization)
8585
}
8686

87-
pub async fn connect(&mut self, connect: Connect) -> Result<(), StateError> {
88-
self.write(Packet::Connect(connect)).await?;
89-
90-
self.flush().await
91-
}
92-
9387
pub async fn flush(&mut self) -> Result<(), crate::state::StateError> {
9488
self.framed
9589
.flush()

rumqttc/src/v5/eventloop.rs

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,7 @@ impl EventLoop {
130130

131131
requests_in_channel.retain(|request| {
132132
match request {
133-
Request::PubAck(_) => false, // Wait for publish retransmission, else the broker could be confused by an unexpected ack
133+
Request::PubAck(_) => false, // Wait for publish retransmission, else the broker could be confused by an unexpected ack
134134
_ => true,
135135
}
136136
});
@@ -398,20 +398,20 @@ async fn mqtt_connect(
398398
options: &mut MqttOptions,
399399
network: &mut Network,
400400
) -> Result<ConnAck, ConnectionError> {
401-
let keep_alive = options.keep_alive().as_secs() as u16;
402-
let clean_start = options.clean_start();
403-
let client_id = options.client_id();
404-
let properties = options.connect_properties();
405-
406-
let connect = Connect {
407-
keep_alive,
408-
client_id,
409-
clean_start,
410-
properties,
411-
};
401+
let packet = Packet::Connect(
402+
Connect {
403+
client_id: options.client_id(),
404+
keep_alive: options.keep_alive().as_secs() as u16,
405+
clean_start: options.clean_start(),
406+
properties: options.connect_properties(),
407+
},
408+
options.last_will(),
409+
options.credentials(),
410+
);
412411

413412
// send mqtt connect packet
414-
network.connect(connect, options).await?;
413+
network.write(packet).await?;
414+
network.flush().await?;
415415

416416
// validate connack
417417
match network.read().await? {

rumqttc/src/v5/framed.rs

Lines changed: 1 addition & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,7 @@ use tokio_util::codec::Framed;
55
use crate::framed::AsyncReadWrite;
66

77
use super::mqttbytes::v5::Packet;
8-
use super::{mqttbytes, Codec, Connect, MqttOptions, MqttState};
9-
use super::{Incoming, StateError};
8+
use super::{mqttbytes, Codec, Incoming, MqttState, StateError};
109

1110
/// Network transforms packets <-> frames efficiently. It takes
1211
/// advantage of pre-allocation, buffering and vectorization when
@@ -86,19 +85,6 @@ impl Network {
8685
.map_err(StateError::Deserialization)
8786
}
8887

89-
pub async fn connect(
90-
&mut self,
91-
connect: Connect,
92-
options: &MqttOptions,
93-
) -> Result<(), StateError> {
94-
let last_will = options.last_will();
95-
let login = options.credentials();
96-
self.write(Packet::Connect(connect, last_will, login))
97-
.await?;
98-
99-
self.flush().await
100-
}
101-
10288
pub async fn flush(&mut self) -> Result<(), StateError> {
10389
self.framed
10490
.flush()

0 commit comments

Comments
 (0)