Refactor control messages

.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

CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [2.0.0] - 2024-04-1x
+
+* Mark `Control` type as `non exhaustive`
+
+* Rename `ControlMessage` to `Control`
+
+* Remove protocol variant services
+
 ## [1.1.0] - 2024-03-07
 
 * Use MqttService::connect_timeout() only for reading protocol version

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "1.1.0"
+version = "2.0.0"
 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"
@@ -16,16 +16,16 @@ features = ["ntex/tokio"]
 
 [dependencies]
 ntex = "1.2"
-bitflags = "2.4"
+bitflags = "2"
 log = "0.4"
 pin-project-lite = "0.2"
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
-thiserror = "1.0"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+thiserror = "1"
 
 [dev-dependencies]
 env_logger = "0.11"
 ntex-tls = "1.1"
 openssl = "0.10"
 test-case = "3.2"
-ntex = { version = "1.2", features = ["tokio", "openssl"] }
+ntex = { version = "1", features = ["tokio", "openssl"] }

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, ControlMessage, ControlResult, MqttServer, Publish, PublishAck, Session,
-};
+use ntex_mqtt::v5::{self, Control, ControlAck, MqttServer, Publish, PublishAck, Session};
 
 #[derive(Clone, Debug)]
 struct MySession {
@@ -70,22 +68,22 @@ async fn publish(
 }
 
 fn control_service_factory() -> impl ServiceFactory<
- ControlMessage<MyServerError>,
+ Control<MyServerError>,
  Session<MySession>,
- Response = ControlResult,
+ Response = ControlAck,
  Error = MyServerError,
  InitError = MyServerError,
 > {
  fn_factory_with_config(|session: Session<MySession>| {
  Ready::Ok(fn_service(move |control| match control {
- v5::ControlMessage::Auth(a) => Ready::Ok(a.ack(v5::codec::Auth::default())),
- v5::ControlMessage::Error(e) => {
+ v5::Control::Auth(a) => Ready::Ok(a.ack(v5::codec::Auth::default())),
+ v5::Control::Error(e) => {
  Ready::Ok(e.ack(v5::codec::DisconnectReasonCode::UnspecifiedError))
  }
- v5::ControlMessage::ProtocolError(e) => Ready::Ok(e.ack()),
- v5::ControlMessage::Ping(p) => Ready::Ok(p.ack()),
- v5::ControlMessage::Disconnect(d) => Ready::Ok(d.ack()),
- v5::ControlMessage::Subscribe(mut s) => {
+ v5::Control::ProtocolError(e) => Ready::Ok(e.ack()),
+ v5::Control::Ping(p) => Ready::Ok(p.ack()),
+ v5::Control::Disconnect(d) => Ready::Ok(d.ack()),
+ v5::Control::Subscribe(mut s) => {
  // store subscribed topics in session, publish service uses this list for echos
  s.iter_mut().for_each(|mut s| {
  session.subscriptions.borrow_mut().push(s.topic().clone());
@@ -94,9 +92,10 @@ fn control_service_factory() -> impl ServiceFactory<
 
  Ready::Ok(s.ack())
  }
- v5::ControlMessage::Unsubscribe(s) => Ready::Ok(s.ack()),
- v5::ControlMessage::Closed(c) => Ready::Ok(c.ack()),
- v5::ControlMessage::PeerGone(c) => Ready::Ok(c.ack()),
+ 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()),
  }))
  })
 }

examples/subs_client.rs

@@ -29,9 +29,9 @@ async fn main() -> std::io::Result<()> {
  let sink = client.sink();
 
  // handle incoming publishes
- ntex::rt::spawn(client.start(fn_service(|control: v5::client::ControlMessage<Error>| {
- match control {
- v5::client::ControlMessage::Publish(publish) => {
+ ntex::rt::spawn(client.start(fn_service(
+ |control: v5::client::Control<Error>| match control {
+ v5::client::Control::Publish(publish) => {
  log::info!(
  "incoming publish: {:?} -> {:?} payload {:?}",
  publish.packet().packet_id,
@@ -40,28 +40,29 @@ async fn main() -> std::io::Result<()> {
  );
  Ready::Ok(publish.ack(v5::codec::PublishAckReason::Success))
  }
- v5::client::ControlMessage::Disconnect(msg) => {
+ v5::client::Control::Disconnect(msg) => {
  log::warn!("Server disconnecting: {:?}", msg);
  Ready::Ok(msg.ack())
  }
- v5::client::ControlMessage::Error(msg) => {
+ v5::client::Control::Error(msg) => {
  log::error!("Codec error: {:?}", msg);
  Ready::Ok(msg.ack(v5::codec::DisconnectReasonCode::UnspecifiedError))
  }
- v5::client::ControlMessage::ProtocolError(msg) => {
+ v5::client::Control::ProtocolError(msg) => {
  log::error!("Protocol error: {:?}", msg);
  Ready::Ok(msg.ack())
  }
- v5::client::ControlMessage::PeerGone(msg) => {
+ v5::client::Control::PeerGone(msg) => {
  log::warn!("Peer closed connection: {:?}", msg.error());
  Ready::Ok(msg.ack())
  }
- v5::client::ControlMessage::Closed(msg) => {
+ v5::client::Control::Closed(msg) => {
  log::warn!("Server closed connection: {:?}", msg);
  Ready::Ok(msg.ack())
  }
- }
- })));
+ _ => Ready::Ok(control.ack()),
+ },
+ )));
 
  // subscribe to topic
  sink.subscribe(None)

src/server.rs

@@ -24,7 +24,7 @@ impl<Err, InitErr>
  InitErr,
  >
 {
- /// Create mqtt protocol selector server
+ /// Create mqtt server
  pub fn new() -> Self {
  MqttServer {
  v3: DefaultProtocolServer::new(ProtocolVersion::MQTT3),
@@ -85,11 +85,8 @@ where
  Error = Err,
  InitError = InitErr,
  > + 'static,
- Cn: ServiceFactory<
- v3::ControlMessage<Err>,
- v3::Session<St>,
- Response = v3::ControlResult,
- > + 'static,
+ Cn: ServiceFactory<v3::Control<Err>, v3::Session<St>, Response = v3::ControlAck>
+ + 'static,
  P: ServiceFactory<v3::Publish, v3::Session<St>, Response = ()> + 'static,
  C::Error: From<Cn::Error>
  + From<Cn::InitError>
@@ -105,28 +102,6 @@ where
  }
  }
 
- /// Service to handle v3 protocol
- pub fn v3_variants(
- self,
- service: v3::Selector<Err, InitErr>,
- ) -> MqttServer<
- impl ServiceFactory<IoBoxed, Response = (), Error = MqttError<Err>, InitError = InitErr>,
- V5,
- Err,
- InitErr,
- >
- where
- Err: 'static,
- InitErr: 'static,
- {
- MqttServer {
- v3: service,
- v5: self.v5,
- connect_timeout: self.connect_timeout,
- _t: marker::PhantomData,
- }
- }
-
  /// Service to handle v5 protocol
  pub fn v5<St, C, Cn, P>(
  self,
@@ -145,11 +120,8 @@ where
  Error = Err,
  InitError = InitErr,
  > + 'static,
- Cn: ServiceFactory<
- v5::ControlMessage<Err>,
- v5::Session<St>,
- Response = v5::ControlResult,
- > + 'static,
+ Cn: ServiceFactory<v5::Control<Err>, v5::Session<St>, Response = v5::ControlAck>
+ + 'static,
  P: ServiceFactory<v5::Publish, v5::Session<St>, Response = v5::PublishAck> + 'static,
  P::Error: fmt::Debug,
  C::Error: From<Cn::Error>
@@ -166,28 +138,6 @@ where
  _t: marker::PhantomData,
  }
  }
-
- /// Service to handle v5 protocol
- pub fn v5_variants<St, C, Cn, P>(
- self,
- service: v5::Selector<Err, InitErr>,
- ) -> MqttServer<
- V3,
- impl ServiceFactory<IoBoxed, Response = (), Error = MqttError<Err>, InitError = InitErr>,
- Err,
- InitErr,
- >
- where
- Err: 'static,
- InitErr: 'static,
- {
- MqttServer {
- v3: self.v3,
- v5: service,
- connect_timeout: self.connect_timeout,
- _t: marker::PhantomData,
- }
- }
 }
 
 impl<V3, V5, Err, InitErr> MqttServer<V3, V5, Err, InitErr>

src/v3/client/connection.rs

@@ -7,11 +7,10 @@
 use ntex::time::{sleep, Millis, Seconds};
 use ntex::util::{Either, Ready};
 
-use crate::error::MqttError;
-use crate::io::Dispatcher;
-use crate::v3::{codec, shared::MqttShared, sink::MqttSink, ControlResult, Publish};
+use crate::v3::{codec, shared::MqttShared, sink::MqttSink, ControlAck, Publish};
+use crate::{error::MqttError, io::Dispatcher};
 
-use super::{control::ControlMessage, dispatcher::create_dispatcher};
+use super::{control::Control, dispatcher::create_dispatcher};
 
 /// Mqtt client
 pub struct Client {
@@ -104,7 +103,7 @@
  self.shared.clone(),
  self.max_receive,
  into_service(|pkt| Ready::Ok(Either::Right(pkt))),
- into_service(|msg: ControlMessage<()>| Ready::<_, ()>::Ok(msg.disconnect())),
+ into_service(|msg: Control<()>| Ready::<_, ()>::Ok(msg.disconnect())),
  );
 
  let _ = Dispatcher::new(self.io, self.shared.clone(), dispatcher, &self.config).await;
@@ -114,8 +113,8 @@
  pub async fn start<F, S, E>(self, service: F) -> Result<(), MqttError<E>>
  where
  E: 'static,
- F: IntoService<S, ControlMessage<E>> + 'static,
- S: Service<ControlMessage<E>, Response = ControlResult, Error = E> + 'static,
+ F: IntoService<S, Control<E>> + 'static,
+ S: Service<Control<E>, Response = ControlAck, Error = E> + 'static,
  {
  if self.keepalive.non_zero() {
  let _ =
@@ -190,7 +189,7 @@
  self.shared.clone(),
  self.max_receive,
  dispatch(self.builder.finish(), self.handlers),
- into_service(|msg: ControlMessage<Err>| Ready::<_, Err>::Ok(msg.disconnect())),
+ into_service(|msg: Control<Err>| Ready::<_, Err>::Ok(msg.disconnect())),
  );
 
  let _ = Dispatcher::new(self.io, self.shared.clone(), dispatcher, &self.config).await;
@@ -199,8 +198,8 @@
  /// Run client and handle control messages
  pub async fn start<F, S>(self, service: F) -> Result<(), MqttError<Err>>
  where
- F: IntoService<S, ControlMessage<Err>>,
- S: Service<ControlMessage<Err>, Response = ControlResult, Error = Err> + 'static,
+ F: IntoService<S, Control<Err>>,
+ S: Service<Control<Err>, Response = ControlAck, Error = Err> + 'static,
  {
  if self.keepalive.non_zero() {
  let _ =