Change handshake timeout behavior

CHANGES.md

@@ -1,5 +1,10 @@
 # Changes
 
+## [0.12.1] - 2023-09-25
+
+* Change handshake timeout behavior (renamed to connect timeout).
+  Timeout handles slow client's Control frame.
+
 ## [0.12.0] - 2023-09-18
 
 * Refactor MqttError type

Cargo.toml

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

src/io.rs

@@ -730,7 +730,7 @@ mod tests {
 
         let (disp, io) =
             Dispatcher::new_debug(nio::Io::new(server), BytesCodec, Srv(counter.clone()));
-        io.encode(Bytes::from_static(b"GET /test HTTP/1\r\n\r\n"), &mut BytesCodec).unwrap();
+        io.encode(Bytes::from_static(b"GET /test HTTP/1\r\n\r\n"), &BytesCodec).unwrap();
         ntex::rt::spawn(async move {
             let _ = disp.await;
         });

src/server.rs

@@ -13,7 +13,7 @@
 pub struct MqttServer<V3, V5, Err, InitErr> {
     v3: V3,
     v5: V5,
-    handshake_timeout: Millis,
+    connect_timeout: Millis,
     _t: marker::PhantomData<(Err, InitErr)>,
 }
 
@@ -30,7 +30,7 @@
         MqttServer {
             v3: DefaultProtocolServer::new(ProtocolVersion::MQTT3),
             v5: DefaultProtocolServer::new(ProtocolVersion::MQTT5),
-            handshake_timeout: Millis(10000),
+            connect_timeout: Millis(10000),
             _t: marker::PhantomData,
         }
     }
@@ -50,12 +50,23 @@
 }
 
 impl<V3, V5, Err, InitErr> MqttServer<V3, V5, Err, InitErr> {
-    /// Set handshake timeout.
+    /// Set client timeout for first `Connect` frame.
+    ///
+    /// Defines a timeout for reading `Connect` frame. If a client does not transmit
+    /// the entire frame within this time, the connection is terminated with
+    /// Mqtt::Handshake(HandshakeError::Timeout) error.
     ///
-    /// Handshake includes `connect` packet.
-    /// By default handshake timeuot is 10 seconds.
+    /// By default, connect timeuot is 10 seconds.
+    pub fn conenct_timeout(mut self, timeout: Seconds) -> Self {
+        self.connect_timeout = timeout.into();
+        self
+    }
+
+    #[deprecated(since = "0.12.1")]
+    #[doc(hidden)]
+    /// Set handshake timeout.
     pub fn handshake_timeout(mut self, timeout: Seconds) -> Self {
-        self.handshake_timeout = timeout.into();
+        self.connect_timeout = timeout.into();
         self
     }
 }
@@ -113,7 +124,7 @@
         MqttServer {
             v3: service.finish(),
             v5: self.v5,
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             _t: marker::PhantomData,
         }
     }
@@ -140,7 +151,7 @@
         MqttServer {
             v3: service,
             v5: self.v5,
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             _t: marker::PhantomData,
         }
     }
@@ -185,7 +196,7 @@
         MqttServer {
             v3: self.v3,
             v5: service.finish(),
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             _t: marker::PhantomData,
         }
     }
@@ -212,7 +223,7 @@
         MqttServer {
             v3: self.v3,
             v5: service,
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             _t: marker::PhantomData,
         }
     }
@@ -241,7 +252,7 @@
         let v5 = v5?;
         Ok(MqttServerImpl {
             handlers: (v3, v5),
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             _t: marker::PhantomData,
         })
     }
@@ -311,7 +322,7 @@
 /// Mqtt Server
 pub struct MqttServerImpl<V3, V5, Err> {
     handlers: (V3, V5),
-    handshake_timeout: Millis,
+    connect_timeout: Millis,
     _t: marker::PhantomData<Err>,
 }
 
@@ -353,7 +364,7 @@
         MqttServerImplResponse {
             ctx,
             state: MqttServerImplState::Version {
-                item: Some((req, VersionCodec, Deadline::new(self.handshake_timeout))),
+                item: Some((req, VersionCodec, Deadline::new(self.connect_timeout))),
             },
             handlers: &self.handlers,
         }

src/topic.rs

@@ -376,8 +376,7 @@ mod tests {
     #[test_case("/finance" => Ok(vec![TopicFilterLevel::Blank, lvl_normal("finance")]) ; "12")]
     #[test_case("finance/" => Ok(vec![lvl_normal("finance"), TopicFilterLevel::Blank]) ; "13")]
     fn parsing(input: &str) -> Result<Vec<TopicFilterLevel>, TopicFilterError> {
-        TopicFilter::try_from(ByteString::from(input))
-            .map(|t| t.levels().iter().cloned().collect())
+        TopicFilter::try_from(ByteString::from(input)).map(|t| t.levels().to_vec())
     }
 
     #[test_case(vec![lvl_normal("sport"), lvl_normal("tennis"), lvl_normal("player1")] => true; "1")]

src/v3/codec/encode.rs

@@ -262,7 +262,7 @@ mod tests {
         let mut v = BytesMut::with_capacity(1024);
         encode(packet, &mut v, get_encoded_size(packet) as u32).unwrap();
         assert_eq!(expected.len(), v.len());
-        assert_eq!(&expected[..], &v[..]);
+        assert_eq!(expected, &v[..]);
     }
 
     #[test]

src/v3/selector.rs

@@ -12,12 +12,10 @@
 use super::shared::{MqttShared, MqttSinkPool};
 use super::{codec as mqtt, MqttServer, Publish, Session};
 
-pub(crate) type SelectItem = (Handshake, Deadline);
-
 type ServerFactory<Err, InitErr> =
-    boxed::BoxServiceFactory<(), SelectItem, Either<SelectItem, ()>, MqttError<Err>, InitErr>;
+    boxed::BoxServiceFactory<(), Handshake, Either<Handshake, ()>, MqttError<Err>, InitErr>;
 
-type Server<Err> = boxed::BoxService<SelectItem, Either<SelectItem, ()>, MqttError<Err>>;
+type Server<Err> = boxed::BoxService<Handshake, Either<Handshake, ()>, MqttError<Err>>;
 
 /// Mqtt server selector
 ///
@@ -26,7 +24,7 @@
 pub struct Selector<Err, InitErr> {
     servers: Vec<ServerFactory<Err, InitErr>>,
     max_size: u32,
-    handshake_timeout: Millis,
+    connect_timeout: Millis,
     pool: Rc<MqttSinkPool>,
     _t: marker::PhantomData<(Err, InitErr)>,
 }
@@ -37,7 +35,7 @@
         Selector {
             servers: Vec::new(),
             max_size: 0,
-            handshake_timeout: Millis(10000),
+            connect_timeout: Millis(10000),
             pool: Default::default(),
             _t: marker::PhantomData,
         }
@@ -49,12 +47,23 @@
     Err: 'static,
     InitErr: 'static,
 {
-    /// Set handshake timeout.
+    /// Set client timeout for first `Connect` frame.
     ///
-    /// Handshake includes `connect` packet and response `connect-ack`.
-    /// By default handshake timeuot is 10 seconds.
+    /// Defines a timeout for reading `Connect` frame. If a client does not transmit
+    /// the entire frame within this time, the connection is terminated with
+    /// Mqtt::Handshake(HandshakeError::Timeout) error.
+    ///
+    /// By default, connect timeuot is 10 seconds.
+    pub fn conenct_timeout(mut self, timeout: Seconds) -> Self {
+        self.connect_timeout = timeout.into();
+        self
+    }
+
+    #[deprecated(since = "0.12.1")]
+    #[doc(hidden)]
+    /// Set handshake timeout.
     pub fn handshake_timeout(mut self, timeout: Seconds) -> Self {
-        self.handshake_timeout = timeout.into();
+        self.connect_timeout = timeout.into();
         self
     }
 
@@ -111,7 +120,7 @@
         Ok(SelectorService {
             servers,
             max_size: self.max_size,
-            handshake_timeout: self.handshake_timeout,
+            connect_timeout: self.connect_timeout,
             pool: self.pool.clone(),
         })
     }
@@ -169,7 +178,7 @@
 pub struct SelectorService<Err> {
     servers: Vec<Server<Err>>,
     max_size: u32,
-    handshake_timeout: Millis,
+    connect_timeout: Millis,
     pool: Rc<MqttSinkPool>,
 }
 
@@ -234,58 +243,11 @@
 
     #[inline]
     fn call<'a>(&'a self, io: IoBoxed, ctx: ServiceCtx<'a, Self>) -> Self::Future<'a> {
-        Box::pin(async move {
-            let codec = mqtt::Codec::default();
-            codec.set_max_size(self.max_size);
-            let shared = Rc::new(MqttShared::new(io.clone(), codec, false, self.pool.clone()));
-            let mut timeout = Deadline::new(self.handshake_timeout);
-
-            // read first packet
-            let result = select(&mut timeout, async {
-                io.recv(&shared.codec)
-                    .await
-                    .map_err(|err| {
-                        log::trace!("Error is received during mqtt handshake: {:?}", err);
-                        MqttError::Handshake(HandshakeError::from(err))
-                    })?
-                    .ok_or_else(|| {
-                        log::trace!("Server mqtt is disconnected during handshake");
-                        MqttError::Handshake(HandshakeError::Disconnected(None))
-                    })
-            })
-            .await;
-
-            let (packet, size) = match result {
-                Either::Left(_) => Err(MqttError::Handshake(HandshakeError::Timeout)),
-                Either::Right(item) => item,
-            }?;
-
-            let connect = match packet {
-                mqtt::Packet::Connect(connect) => connect,
-                packet => {
-                    log::info!("MQTT-3.1.0-1: Expected CONNECT packet, received {:?}", packet);
-                    return Err(MqttError::Handshake(HandshakeError::Protocol(
-                        ProtocolError::unexpected_packet(
-                            packet.packet_type(),
-                            "Expected CONNECT packet [MQTT-3.1.0-1]",
-                        ),
-                    )));
-                }
-            };
-
-            // call servers
-            let mut item = (Handshake::new(connect, size, io, shared), timeout);
-            for srv in &self.servers {
-                match ctx.call(srv, item).await? {
-                    Either::Left(result) => {
-                        item = result;
-                    }
-                    Either::Right(_) => return Ok(()),
-                }
-            }
-            log::error!("Cannot handle CONNECT packet {:?}", item.0);
-            Err(MqttError::Handshake(HandshakeError::Server("Cannot handle CONNECT packet")))
-        })
+        Service::<(IoBoxed, Deadline)>::call(
+            self,
+            (io, Deadline::new(self.connect_timeout)),
+            ctx,
+        )
     }
 }
 
@@ -353,7 +315,7 @@
             };
 
             // call servers
-            let mut item = (Handshake::new(connect, size, io, shared), timeout);
+            let mut item = Handshake::new(connect, size, io, shared);
             for srv in &self.servers {
                 match ctx.call(srv, item).await? {
                     Either::Left(result) => {
@@ -362,7 +324,7 @@
                     Either::Right(_) => return Ok(()),
                 }
             }
-            log::error!("Cannot handle CONNECT packet {:?}", item.0.packet());
+            log::error!("Cannot handle CONNECT packet {:?}", item.packet());
             Err(MqttError::Handshake(HandshakeError::Server("Cannot handle CONNECT packet")))
         })
     }