Fix disconnect packet handling

CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.8.9] - 2022-09-16
+
+* v3: Send disconnect packet on sink close
+
+* v3: Treat disconnect packet as error on client side
+
 ## [0.8.8] - 2022-08-22
 
 * Allow to get inner io stream and codec for negotiated clients

Cargo.toml

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

src/v3/client/connector.rs

@@ -229,7 +229,7 @@ where
                     ClientError::Disconnected(None)
                 })?;
 
-            let shared = Rc::new(MqttShared::new(io.get_ref(), codec, max_send, pool));
+            let shared = Rc::new(MqttShared::new(io.get_ref(), codec, max_send, true, pool));
 
             match packet {
                 codec::Packet::ConnectAck { session_present, return_code } => {

src/v3/client/control.rs

@@ -10,6 +10,8 @@ pub enum ControlMessage<E> {
     /// Unhandled publish packet
     Publish(Publish),
     /// Disconnect packet
+    #[doc(hidden)]
+    #[deprecated(since = "0.8.9", note = "Disconnect is not allowed on client side")]
     Disconnect(Disconnect),
     /// Connection closed
     Closed(Closed),
@@ -26,10 +28,6 @@ impl<E> ControlMessage<E> {
         ControlMessage::Publish(Publish(pkt))
     }
 
-    pub(super) fn dis() -> Self {
-        ControlMessage::Disconnect(Disconnect)
-    }
-
     pub(super) fn closed(is_error: bool) -> Self {
         ControlMessage::Closed(Closed::new(is_error))
     }
@@ -46,6 +44,7 @@ impl<E> ControlMessage<E> {
         ControlMessage::PeerGone(PeerGone(err))
     }
 
+    /// Initiate clean disconnect
     pub fn disconnect(&self) -> ControlResult {
         ControlResult { result: ControlResultKind::Disconnect }
     }

src/v3/client/dispatcher.rs

@@ -138,9 +138,15 @@ where
             DispatchItem::Item(codec::Packet::PingRequest) => {
                 Either::Right(Either::Left(Ready::Ok(Some(codec::Packet::PingResponse))))
             }
-            DispatchItem::Item(codec::Packet::Disconnect) => Either::Right(Either::Right(
-                ControlResponse::new(ControlMessage::dis(), &self.inner),
-            )),
+            DispatchItem::Item(codec::Packet::Disconnect) => {
+                Either::Right(Either::Left(Ready::Err(
+                    ProtocolError::Unexpected(
+                        packet_type::DISCONNECT,
+                        "Disconnect packet is not allowed",
+                    )
+                    .into(),
+                )))
+            }
             DispatchItem::Item(codec::Packet::SubscribeAck { packet_id, status }) => {
                 if let Err(e) = self.sink.pkt_ack(Ack::Subscribe { packet_id, status }) {
                     Either::Right(Either::Left(Ready::Err(MqttError::Protocol(e))))
@@ -159,7 +165,7 @@ where
                 Either::Right(Either::Left(Ready::Err(
                     ProtocolError::Unexpected(
                         packet_type::SUBSCRIBE,
-                        "Subscribe packet is not supported",
+                        "Subscribe packet is not allowed",
                     )
                     .into(),
                 )))
@@ -168,7 +174,7 @@ where
                 Either::Right(Either::Left(Ready::Err(
                     ProtocolError::Unexpected(
                         packet_type::UNSUBSCRIBE,
-                        "Unsubscribe packet is not supported",
+                        "Unsubscribe packet is not allowed",
                     )
                     .into(),
                 )))
@@ -302,7 +308,7 @@ where
                 ControlResultKind::Unsubscribe(_) => unreachable!(),
                 ControlResultKind::Disconnect => {
                     this.inner.sink.close();
-                    Some(codec::Packet::Disconnect)
+                    None
                 }
                 ControlResultKind::Closed | ControlResultKind::Nothing => None,
             },

src/v3/selector.rs

@@ -241,6 +241,7 @@ where
             io.clone(),
             mqtt::Codec::default().max_size(self.max_size),
             16,
+            false,
             self.pool.clone(),
         ));
         let mut timeout = Deadline::new(self.handshake_timeout);
@@ -318,6 +319,7 @@ where
             io.get_ref(),
             mqtt::Codec::default().max_size(self.max_size),
             16,
+            false,
             self.pool.clone(),
         ));
 

src/v3/server.rs

@@ -321,6 +321,7 @@ where
             io.get_ref(),
             mqtt::Codec::default().max_size(self.max_size),
             16,
+            false,
             self.pool.clone(),
         ));
         let handshake_timeout = self.handshake_timeout;

src/v3/shared.rs

@@ -44,6 +44,7 @@ pub struct MqttShared {
     pub(super) inflight_idx: Cell<u16>,
     pub(super) pool: Rc<MqttSinkPool>,
     pub(super) codec: codec::Codec,
+    pub(super) client: bool,
 }
 
 pub(super) struct MqttSharedQueues {
@@ -57,12 +58,14 @@ impl MqttShared {
         io: IoRef,
         codec: codec::Codec,
         cap: usize,
+        client: bool,
         pool: Rc<MqttSinkPool>,
     ) -> Self {
         Self {
             io,
             pool,
             codec,
+            client,
             cap: Cell::new(cap),
             queues: RefCell::new(MqttSharedQueues {
                 inflight: HashMap::default(),

src/v3/sink.rs

@@ -55,6 +55,9 @@ impl MqttSink {
     #[inline]
     /// Close mqtt connection
     pub fn close(&self) {
+        if self.0.client {
+            let _ = self.0.io.encode(codec::Packet::Disconnect, &self.0.codec);
+        }
         self.0.io.close();
         self.0.with_queues(|q| {
             q.inflight.clear();