Drop unneeded HandshakeError::Server

ntex-rs · Sep 25, 2023 · 5b29020 · 5b29020
1 parent 2e6fc91
commit 5b29020
Show file tree

Hide file tree

Showing 10 changed files with 42 additions and 27 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -1,5 +1,9 @@
 # Changes
 
+## [0.12.2] - 2023-09-25
+
+* Drop unneeded HandshakeError::Server
+
 ## [0.12.1] - 2023-09-25
 
 * Change handshake timeout behavior (renamed to connect timeout).

diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "0.12.1"
+version = "0.12.2"
 authors = ["ntex contributors <[email protected]>"]
 description = "Client and Server framework for MQTT v5 and v3.1.1 protocols"
 documentation = "https://docs.rs/ntex-mqtt"

diff --git a/examples/mqtt-ws-server.rs b/examples/mqtt-ws-server.rs
@@ -6,7 +6,7 @@ use ntex::io::{Filter, Io};
 use ntex::service::{chain_factory, ServiceFactory};
 use ntex::util::{variant, Ready};
 use ntex::ws;
-use ntex_mqtt::{v3, v5, HandshakeError, MqttError, MqttServer};
+use ntex_mqtt::{v3, v5, HandshakeError, MqttError, MqttServer, ProtocolError};
 use ntex_tls::openssl::Acceptor;
 use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
 
@@ -101,9 +101,11 @@ async fn main() -> std::io::Result<()> {
                         return match result {
                             Some(Protocol::Mqtt) => Ok(variant::Variant2::V1(io)),
                             Some(Protocol::Http) => Ok(variant::Variant2::V2(io)),
-                            Some(Protocol::Unknown) => Err(MqttError::Handshake(
-                                HandshakeError::Server("Unsupported protocol"),
-                            )),
+                            Some(Protocol::Unknown) => {
+                                Err(MqttError::Handshake(HandshakeError::Protocol(
+                                    ProtocolError::generic_violation("Unsupported protocol"),
+                                )))
+                            }
                             None => {
                                 // need to read more data
                                 io.read_ready().await?;
@@ -140,8 +142,10 @@ async fn main() -> std::io::Result<()> {
                                                     )
                                                     .await?;
                                                     return Err(MqttError::Handshake(
-                                                        HandshakeError::Server(
-                                                            "WebSockets handshake error",
+                                                        HandshakeError::Protocol(
+                                                            ProtocolError::generic_violation(
+                                                                "WebSockets handshake error",
+                                                            ),
                                                         ),
                                                     ));
                                                 }
@@ -178,8 +182,8 @@ async fn main() -> std::io::Result<()> {
                             // adapt service error to mqtt error
                             .map_err(|e| {
                                 log::info!("Http server error: {:?}", e);
-                                MqttError::Handshake(HandshakeError::Server(
-                                    "Http server error",
+                                MqttError::Handshake(HandshakeError::Protocol(
+                                    ProtocolError::generic_violation("Http server error"),
                                 ))
                             })),
                 )

diff --git a/src/error.rs b/src/error.rs
@@ -30,9 +30,6 @@ pub enum HandshakeError<E> {
     /// Peer disconnect
     #[error("Peer is disconnected, error: {:?}", _0)]
     Disconnected(Option<io::Error>),
-    /// Server error
-    #[error("Server error: {}", _0)]
-    Server(&'static str),
 }
 
 /// Protocol level errors
@@ -81,7 +78,7 @@ impl ProtocolError {
             inner: ViolationInner::Common { reason, message },
         })
     }
-    pub(crate) fn generic_violation(message: &'static str) -> Self {
+    pub fn generic_violation(message: &'static str) -> Self {
         Self::violation(DisconnectReasonCode::ProtocolError, message)
     }
 

diff --git a/src/lib.rs b/src/lib.rs
@@ -19,7 +19,7 @@ mod session;
 mod types;
 mod version;
 
-pub use self::error::{HandshakeError, MqttError};
+pub use self::error::{HandshakeError, MqttError, ProtocolError};
 pub use self::server::MqttServer;
 pub use self::session::Session;
 pub use self::topic::{TopicFilter, TopicFilterError, TopicFilterLevel};

diff --git a/src/v3/client/dispatcher.rs b/src/v3/client/dispatcher.rs
@@ -120,7 +120,8 @@ where
                     if !inner.inflight.borrow_mut().insert(pid) {
                         log::trace!("Duplicated packet id for publish packet: {:?}", pid);
                         return Either::Right(Either::Left(Ready::Err(MqttError::Handshake(
-                            HandshakeError::Server("Duplicated packet id for publish packet"),
+                            HandshakeError::Protocol(
+                                ProtocolError::generic_violation("PUBLISH received with packet id that is already in use [MQTT-2.2.1-3]"))
                         ))));
                     }
                 }
@@ -382,8 +383,9 @@ mod tests {
         ))));
         let err = f.await.err().unwrap();
         match err {
-            MqttError::Handshake(HandshakeError::Server(msg)) => {
-                assert!(msg == "Duplicated packet id for publish packet")
+            MqttError::Handshake(HandshakeError::Protocol(msg)) => {
+                assert!(format!("{}", msg)
+                    .contains("PUBLISH received with packet id that is already in use"))
             }
             _ => panic!(),
         }

diff --git a/src/v3/dispatcher.rs b/src/v3/dispatcher.rs
@@ -60,9 +60,9 @@ where
                 BufferServiceError::Service(e) => {
                     MqttError::Handshake(HandshakeError::Service(E::from(e)))
                 }
-                BufferServiceError::RequestCanceled => MqttError::Handshake(
-                    HandshakeError::Server("Request handling has been canceled"),
-                ),
+                BufferServiceError::RequestCanceled => {
+                    MqttError::Handshake(HandshakeError::Disconnected(None))
+                }
             });
 
             Ok(

diff --git a/src/v3/selector.rs b/src/v3/selector.rs
@@ -1,4 +1,4 @@
-use std::{fmt, future::Future, marker, rc::Rc, task::Context, task::Poll};
+use std::{fmt, future::Future, io, marker, rc::Rc, task::Context, task::Poll};
 
 use ntex::io::{Filter, Io, IoBoxed};
 use ntex::service::{boxed, Service, ServiceCtx, ServiceFactory};
@@ -325,7 +325,10 @@ where
                 }
             }
             log::error!("Cannot handle CONNECT packet {:?}", item.packet());
-            Err(MqttError::Handshake(HandshakeError::Server("Cannot handle CONNECT packet")))
+            Err(MqttError::Handshake(HandshakeError::Disconnected(Some(io::Error::new(
+                io::ErrorKind::Other,
+                "Cannot handle CONNECT packet",
+            )))))
         })
     }
 }
diff --git a/src/v5/dispatcher.rs b/src/v5/dispatcher.rs
@@ -58,9 +58,9 @@ where
                 BufferServiceError::Service(e) => {
                     MqttError::Handshake(HandshakeError::Service(E::from(e)))
                 }
-                BufferServiceError::RequestCanceled => MqttError::Handshake(
-                    HandshakeError::Server("Request handling has been canceled"),
-                ),
+                BufferServiceError::RequestCanceled => {
+                    MqttError::Handshake(HandshakeError::Disconnected(None))
+                }
             });
 
             Ok(crate::inflight::InFlightService::new(

diff --git a/src/v5/selector.rs b/src/v5/selector.rs
@@ -1,4 +1,6 @@
-use std::{convert::TryFrom, fmt, future::Future, marker, rc::Rc, task::Context, task::Poll};
+use std::{
+    convert::TryFrom, fmt, future::Future, io, marker, rc::Rc, task::Context, task::Poll,
+};
 
 use ntex::io::{Filter, Io, IoBoxed};
 use ntex::service::{boxed, Service, ServiceCtx, ServiceFactory};
@@ -328,7 +330,10 @@ where
                 }
             }
             log::error!("Cannot handle CONNECT packet {:?}", item);
-            Err(MqttError::Handshake(HandshakeError::Server("Cannot handle CONNECT packet")))
+            Err(MqttError::Handshake(HandshakeError::Disconnected(Some(io::Error::new(
+                io::ErrorKind::Other,
+                "Cannot handle CONNECT packet",
+            )))))
         })
     }
 }