Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Reorganize ConnectionError #370

Merged
merged 3 commits into from
Mar 14, 2022
Merged

Reorganize ConnectionError #370

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c6acd4f
In case of connetion failure, return error code
de-sh Mar 11, 2022
10172c7
New variant NotConnAck
de-sh Mar 13, 2022
7f9d4f9
clean up
de-sh Mar 13, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 16 additions & 26 deletions rumqttc/src/eventloop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ use crate::{framed::Network, Transport};
use crate::{tls, Incoming, MqttState, Packet, Request, StateError};
use crate::{MqttOptions, Outgoing};

use crate::mqttbytes;
use crate::mqttbytes::v4::*;
use async_channel::{bounded, Receiver, Sender};
#[cfg(feature = "websocket")]
Expand All @@ -29,14 +28,17 @@ pub enum ConnectionError {
MqttState(#[from] StateError),
#[error("Timeout")]
Timeout(#[from] Elapsed),
#[error("Packet parsing error: {0}")]
Mqtt4Bytes(mqttbytes::Error),
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This variant is not being used anywhere, hence decided to get it removed.

#[error("Network: {0}")]
Network(#[from] tls::Error),
#[cfg(feature = "websocket")]
#[error("Websocket: {0}")]
Websocket(#[from] async_tungstenite::tungstenite::error::Error),
#[error("TLS: {0}")]
Tls(#[from] tls::Error),
#[error("I/O: {0}")]
Io(#[from] io::Error),
#[error("Stream done")]
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here

StreamDone,
#[error("Connection refused, return code: {0:?}")]
ConnectionRefused(ConnectReturnCode),
#[error("Expected ConnAck packet, received: {0:?}")]
NotConnAck(Packet),
#[error("Requests done")]
RequestsDone,
#[error("Cancel request by the user")]
Expand Down Expand Up @@ -293,9 +295,7 @@ async fn network_connect(options: &MqttOptions) -> Result<Network, ConnectionErr
.body(())
.unwrap();

let (socket, _) = connect_async(request)
.await
.map_err(|e| io::Error::new(io::ErrorKind::ConnectionRefused, e))?;
let (socket, _) = connect_async(request).await?;

Network::new(WsStream::new(socket), options.max_incoming_packet_size)
}
Expand All @@ -310,9 +310,7 @@ async fn network_connect(options: &MqttOptions) -> Result<Network, ConnectionErr

let connector = tls::tls_connector(&tls_config).await?;

let (socket, _) = connect_async_with_tls_connector(request, Some(connector))
.await
.map_err(|e| io::Error::new(io::ErrorKind::ConnectionRefused, e))?;
let (socket, _) = connect_async_with_tls_connector(request, Some(connector)).await?;

Network::new(WsStream::new(socket), options.max_incoming_packet_size)
}
Expand Down Expand Up @@ -348,21 +346,13 @@ async fn mqtt_connect(

// wait for 'timeout' time to validate connack
let packet = time::timeout(Duration::from_secs(options.connection_timeout()), async {
let packet = match network.read().await? {
match network.read().await? {
Incoming::ConnAck(connack) if connack.code == ConnectReturnCode::Success => {
Packet::ConnAck(connack)
Ok(Packet::ConnAck(connack))
}
Incoming::ConnAck(connack) => {
let error = format!("Broker rejected. Reason = {:?}", connack.code);
return Err(io::Error::new(io::ErrorKind::InvalidData, error));
}
packet => {
let error = format!("Expecting connack. Received = {:?}", packet);
return Err(io::Error::new(io::ErrorKind::InvalidData, error));
}
};

io::Result::Ok(packet)
Incoming::ConnAck(connack) => Err(ConnectionError::ConnectionRefused(connack.code)),
packet => Err(ConnectionError::NotConnAck(packet)),
}
})
.await??;

Expand Down
16 changes: 9 additions & 7 deletions rumqttc/tests/reliability.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,16 +426,18 @@ async fn next_poll_after_connect_failure_reconnects() {
time::sleep(Duration::from_secs(1)).await;
let mut eventloop = EventLoop::new(options, 5);

let event = eventloop.poll().await;
let error = "Broker rejected. Reason = BadUserNamePassword";
match event {
Err(ConnectionError::Io(e)) => assert_eq!(e.to_string(), error),
match eventloop.poll().await {
Err(ConnectionError::ConnectionRefused(ConnectReturnCode::BadUserNamePassword)) => (),
v => panic!("Expected bad username password error. Found = {:?}", v),
}

let event = eventloop.poll().await.unwrap();
let connack = ConnAck::new(ConnectReturnCode::Success, false);
assert_eq!(event, Event::Incoming(Packet::ConnAck(connack)));
match eventloop.poll().await {
Ok(Event::Incoming(Packet::ConnAck(ConnAck {
code: ConnectReturnCode::Success,
session_present: false,
}))) => (),
v => panic!("Expected ConnAck Success. Found = {:?}", v),
}
}

#[tokio::test]
Expand Down