Mark Control as non exhaustive

ntex-rs · Apr 15, 2024 · cc69135 · cc69135
1 parent 2f7bbc5
commit cc69135
Show file tree

Hide file tree

Showing 11 changed files with 102 additions and 17 deletions.
diff --git a/.github/workflows/cov.yml b/.github/workflows/cov.yml
@@ -28,7 +28,8 @@ jobs:
         run: cargo llvm-cov --all-features --workspace --lcov --output-path lcov.info
 
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
+          token: ${{ secrets.CODECOV_TOKEN }}
           files: lcov.info
           fail_ci_if_error: true
diff --git a/CHANGES.md b/CHANGES.md
@@ -2,6 +2,8 @@
 
 ## [2.0.0] - 2024-04-1x
 
+* Mark `Control` type as `non exhaustive`
+
 * Rename `ControlMessage` to `Control`
 
 * Remove protocol variant services

diff --git a/examples/subs.rs b/examples/subs.rs
@@ -2,9 +2,7 @@ use std::cell::RefCell;
 
 use ntex::service::{fn_factory_with_config, fn_service, ServiceFactory};
 use ntex::util::{ByteString, Ready};
-use ntex_mqtt::v5::{
-    self, Control, ControlAck, MqttServer, Publish, PublishAck, Session,
-};
+use ntex_mqtt::v5::{self, Control, ControlAck, MqttServer, Publish, PublishAck, Session};
 
 #[derive(Clone, Debug)]
 struct MySession {
@@ -97,6 +95,7 @@ fn control_service_factory() -> impl ServiceFactory<
             v5::Control::Unsubscribe(s) => Ready::Ok(s.ack()),
             v5::Control::Closed(c) => Ready::Ok(c.ack()),
             v5::Control::PeerGone(c) => Ready::Ok(c.ack()),
+            _ => Ready::Ok(control.ack()),
         }))
     })
 }

diff --git a/examples/subs_client.rs b/examples/subs_client.rs
@@ -29,8 +29,8 @@ async fn main() -> std::io::Result<()> {
     let sink = client.sink();
 
     // handle incoming publishes
-    ntex::rt::spawn(client.start(fn_service(|control: v5::client::Control<Error>| {
-        match control {
+    ntex::rt::spawn(client.start(fn_service(
+        |control: v5::client::Control<Error>| match control {
             v5::client::Control::Publish(publish) => {
                 log::info!(
                     "incoming publish: {:?} -> {:?} payload {:?}",
@@ -60,8 +60,9 @@ async fn main() -> std::io::Result<()> {
                 log::warn!("Server closed connection: {:?}", msg);
                 Ready::Ok(msg.ack())
             }
-        }
-    })));
+            _ => Ready::Ok(control.ack()),
+        },
+    )));
 
     // subscribe to topic
     sink.subscribe(None)

diff --git a/src/v3/client/control.rs b/src/v3/client/control.rs
@@ -3,6 +3,8 @@ use std::io;
 pub use crate::v3::control::{Closed, ControlAck, Disconnect, Error, PeerGone, ProtocolError};
 use crate::v3::{codec, control::ControlAckKind, error};
 
+/// Client control messages
+#[non_exhaustive]
 #[derive(Debug)]
 pub enum Control<E> {
     /// Unhandled publish packet
@@ -42,6 +44,17 @@ impl<E> Control<E> {
     pub fn disconnect(&self) -> ControlAck {
         ControlAck { result: ControlAckKind::Disconnect }
     }
+
+    /// Ack control message
+    pub fn ack(self) -> ControlAck {
+        match self {
+            Control::Publish(msg) => msg.ack(),
+            Control::Closed(msg) => msg.ack(),
+            Control::Error(msg) => msg.ack(),
+            Control::ProtocolError(msg) => msg.ack(),
+            Control::PeerGone(msg) => msg.ack(),
+        }
+    }
 }
 
 #[derive(Debug)]

diff --git a/src/v3/control.rs b/src/v3/control.rs
@@ -4,6 +4,8 @@ use std::{io, marker::PhantomData, num::NonZeroU16};
 use super::codec;
 use crate::{error, types::QoS};
 
+/// Server control messages
+#[non_exhaustive]
 #[derive(Debug)]
 pub enum Control<E> {
     /// Ping packet
@@ -86,6 +88,26 @@ impl<E> Control<E> {
     pub fn disconnect(&self) -> ControlAck {
         ControlAck { result: ControlAckKind::Disconnect }
     }
+
+    /// Ack control message
+    pub fn ack(self) -> ControlAck {
+        match self {
+            Control::Ping(msg) => msg.ack(),
+            Control::Disconnect(msg) => msg.ack(),
+            Control::Subscribe(_) => {
+                log::warn!("Subscribe is not supported");
+                ControlAck { result: ControlAckKind::Disconnect }
+            }
+            Control::Unsubscribe(_) => {
+                log::warn!("Unsubscribe is not supported");
+                ControlAck { result: ControlAckKind::Disconnect }
+            }
+            Control::Closed(msg) => msg.ack(),
+            Control::Error(msg) => msg.ack(),
+            Control::ProtocolError(msg) => msg.ack(),
+            Control::PeerGone(msg) => msg.ack(),
+        }
+    }
 }
 
 #[derive(Copy, Clone, Debug)]

diff --git a/src/v5/client/control.rs b/src/v5/client/control.rs
@@ -6,6 +6,8 @@ use crate::{error, v5::codec};
 
 pub use crate::v5::control::{Closed, ControlAck, Disconnect, Error, ProtocolError};
 
+/// Client control messages
+#[non_exhaustive]
 #[derive(Debug)]
 pub enum Control<E> {
     /// Unhandled publish packet
@@ -50,6 +52,22 @@ impl<E> Control<E> {
     pub fn disconnect(&self, pkt: codec::Disconnect) -> ControlAck {
         ControlAck { packet: Some(codec::Packet::Disconnect(pkt)), disconnect: true }
     }
+
+    /// Ack control message
+    pub fn ack(self) -> ControlAck {
+        match self {
+            Control::Publish(_) => {
+                crate::v5::disconnect("Publish control message is not supported")
+            }
+            Control::Disconnect(msg) => msg.ack(),
+            Control::Closed(msg) => msg.ack(),
+            Control::Error(_) => {
+                crate::v5::disconnect("Error control message is not supported")
+            }
+            Control::ProtocolError(msg) => msg.ack(),
+            Control::PeerGone(msg) => msg.ack(),
+        }
+    }
 }
 
 #[derive(Debug)]

diff --git a/src/v5/control.rs b/src/v5/control.rs
@@ -5,7 +5,8 @@ use ntex::util::ByteString;
 use super::codec::{self, DisconnectReasonCode, QoS, UserProperties};
 use crate::error;
 
-/// Control messages
+/// Server control messages
+#[non_exhaustive]
 #[derive(Debug)]
 pub enum Control<E> {
     /// Auth packet from a client
@@ -100,6 +101,21 @@ impl<E> Control<E> {
     pub fn disconnect_with(&self, pkt: codec::Disconnect) -> ControlAck {
         ControlAck { packet: Some(codec::Packet::Disconnect(pkt)), disconnect: true }
     }
+
+    /// Ack control message
+    pub fn ack(self) -> ControlAck {
+        match self {
+            Control::Auth(_) => super::disconnect("Auth control message is not supported"),
+            Control::Ping(msg) => msg.ack(),
+            Control::Disconnect(msg) => msg.ack(),
+            Control::Subscribe(msg) => msg.ack(),
+            Control::Unsubscribe(msg) => msg.ack(),
+            Control::Closed(msg) => msg.ack(),
+            Control::Error(_) => super::disconnect("Error control message is not supported"),
+            Control::ProtocolError(msg) => msg.ack(),
+            Control::PeerGone(msg) => msg.ack(),
+        }
+    }
 }
 
 #[derive(Debug)]

diff --git a/src/v5/mod.rs b/src/v5/mod.rs
@@ -28,3 +28,15 @@ pub use crate::topic::{TopicFilter, TopicFilterError};
 pub use crate::types::QoS;
 
 const RECEIVE_MAX_DEFAULT: NonZeroU16 = unsafe { NonZeroU16::new_unchecked(65_535) };
+
+fn disconnect(msg: &'static str) -> ControlAck {
+    log::error!("{}", msg);
+
+    ControlAck {
+        packet: Some(
+            codec::Disconnect::new(codec::DisconnectReasonCode::ImplementationSpecificError)
+                .into(),
+        ),
+        disconnect: true,
+    }
+}
diff --git a/tests/test_server.rs b/tests/test_server.rs
@@ -449,9 +449,10 @@ async fn handle_or_drop_publish_after_disconnect(
     .unwrap();
     io.encode(codec::Packet::Disconnect, &codec).unwrap();
     io.flush(true).await.unwrap();
+    sleep(Millis(350)).await;
+    io.close();
     drop(io);
-
-    sleep(Millis(50)).await;
+    sleep(Millis(250)).await;
 
     assert!(disconnect.load(Relaxed));
 
@@ -770,7 +771,7 @@ async fn test_frame_read_rate() -> std::io::Result<()> {
     sleep(Millis(1000)).await;
     assert!(!check.load(Relaxed));
 
-    sleep(Millis(2100)).await;
+    sleep(Millis(2300)).await;
     assert!(check.load(Relaxed));
 
     Ok(())

diff --git a/tests/test_server_v5.rs b/tests/test_server_v5.rs
@@ -7,8 +7,8 @@ use ntex::util::{lazy, ByteString, Bytes, BytesMut, Ready};
 use ntex::{codec::Encoder, server, service::fn_service};
 
 use ntex_mqtt::v5::{
-    client, codec, error, Control, Handshake, HandshakeAck, MqttServer, Publish,
-    PublishAck, QoS, Session,
+    client, codec, error, Control, Handshake, HandshakeAck, MqttServer, Publish, PublishAck,
+    QoS, Session,
 };
 
 struct St;
@@ -1070,9 +1070,9 @@ async fn handle_or_drop_publish_after_disconnect(
     )
     .unwrap();
     io.flush(true).await.unwrap();
+    io.close();
     drop(io);
-
-    sleep(Millis(50)).await;
+    sleep(Millis(250)).await;
 
     assert!(disconnect.load(Relaxed));
 
@@ -1297,7 +1297,7 @@ async fn test_frame_read_rate() -> std::io::Result<()> {
     sleep(Millis(1000)).await;
     assert!(!check.load(Relaxed));
 
-    sleep(Millis(2100)).await;
+    sleep(Millis(2300)).await;
     assert!(check.load(Relaxed));
 
     Ok(())