Remove inflight limit for client control service (#110)

CHANGES.md

@@ -4,6 +4,10 @@
 
 * Allow to get inner io stream and codec for negotiated clients
 
+* Remove inflight limit for client's control service
+
+* v3: Add Debug trait for client's ControlMessage
+
 ## [0.8.7] - 2022-06-09
 
 * v5: Encoding missing will properties: will_delay_interval_sec, is_utf8_payload, message_expiry_interval, content_type, response_topic, correlation_data, user_properties

src/v3/client/control.rs

@@ -5,6 +5,7 @@ pub use crate::v3::control::{
 };
 use crate::v3::{codec, control::ControlResultKind, error};
 
+#[derive(Debug)]
 pub enum ControlMessage<E> {
     /// Unhandled publish packet
     Publish(Publish),
@@ -50,6 +51,7 @@ impl<E> ControlMessage<E> {
     }
 }
 
+#[derive(Debug)]
 pub struct Publish(codec::Publish);
 
 impl Publish {

src/v3/client/dispatcher.rs

@@ -4,7 +4,7 @@ use std::{future::Future, marker::PhantomData, num::NonZeroU16, pin::Pin, rc::Rc
 
 use ntex::io::DispatchItem;
 use ntex::service::Service;
-use ntex::util::{buffer::BufferService, inflight::InFlightService, Either, HashSet, Ready};
+use ntex::util::{inflight::InFlightService, Either, HashSet, Ready};
 
 use crate::v3::shared::{Ack, MqttShared};
 use crate::v3::{codec, control::ControlResultKind, publish::Publish, sink::MqttSink};
@@ -24,16 +24,11 @@ where
     T: Service<Publish, Response = Either<(), Publish>, Error = E> + 'static,
     C: Service<ControlMessage<E>, Response = ControlResult, Error = E> + 'static,
 {
-    // limit inflight control messages
-    let control = BufferService::new(
-        16,
-        || MqttError::<E>::Disconnected(None),
-        // limit number of in-flight messages
-        InFlightService::new(1, control.map_err(MqttError::Service)),
-    );
-
     // limit number of in-flight messages
-    InFlightService::new(inflight, Dispatcher::new(sink, publish, control))
+    InFlightService::new(
+        inflight,
+        Dispatcher::new(sink, publish, control.map_err(MqttError::Service)),
+    )
 }
 
 /// Mqtt protocol dispatcher

src/v5/client/dispatcher.rs

@@ -4,7 +4,7 @@ use std::{future::Future, marker::PhantomData, num::NonZeroU16, pin::Pin, rc::Rc
 
 use ntex::io::DispatchItem;
 use ntex::service::Service;
-use ntex::util::{buffer::BufferService, inflight::InFlightService, Either, HashSet, Ready};
+use ntex::util::{Either, HashSet, Ready};
 
 use crate::error::{MqttError, ProtocolError};
 use crate::types::packet_type;
@@ -26,14 +26,13 @@ where
     T: Service<Publish, Response = Either<Publish, PublishAck>, Error = E> + 'static,
     C: Service<ControlMessage<E>, Response = ControlResult, Error = E> + 'static,
 {
-    let control = BufferService::new(
-        16,
-        || MqttError::<C::Error>::Disconnected(None),
-        // limit number of in-flight messages
-        InFlightService::new(1, control.map_err(MqttError::Service)),
-    );
-
-    Dispatcher::<_, _, E>::new(sink, max_receive as usize, max_topic_alias, publish, control)
+    Dispatcher::<_, _, E>::new(
+        sink,
+        max_receive as usize,
+        max_topic_alias,
+        publish,
+        control.map_err(MqttError::Service),
+    )
 }
 
 /// Mqtt protocol dispatcher