Implement receiving and sending QoS 2 messages

Author: rmqtt-rs

src/server.rs

@@ -165,16 +165,16 @@ where
             > + 'static,
         P: ServiceFactory<
                 Config = v5::Session<St>,
-                Request = v5::Publish,
-                Response = v5::PublishAck,
+                Request = v5::PublishMessage,
+                Response = v5::PublishResult,
             > + 'static,
         P::Error: fmt::Debug,
         C::Error: From<Cn::Error>
             + From<Cn::InitError>
             + From<P::Error>
             + From<P::InitError>
             + fmt::Debug,
-        v5::PublishAck: TryFrom<P::Error, Error = C::Error>,
+        v5::PublishResult: TryFrom<P::Error, Error = C::Error>,
     {
         MqttServer {
             v3: self.v3,

src/topic.rs

@@ -489,4 +489,3 @@ mod tests {
         assert!(Topic::from_str("$SYS/monitor/+").unwrap().matches_str("$SYS/monitor/Clients"));
     }
 }
-

src/v5/client/dispatcher.rs

@@ -161,7 +161,7 @@ where
 
                         // check for duplicated packet id
                         if !inner.inflight.insert(pid) {
-                            self.inner.sink.send(codec::Packet::PublishAck(
+                            let _ = self.inner.sink.send(codec::Packet::PublishAck(
                                 codec::PublishAck {
                                     packet_id: pid,
                                     reason_code: codec::PublishAckReason::PacketIdentifierInUse,
@@ -453,7 +453,7 @@ where
 
         if self.error {
             if let Some(pkt) = result.packet {
-                self.inner.sink.send(pkt)
+                let _ = self.inner.sink.send(pkt);
             }
             if result.disconnect {
                 self.inner.sink.drop_sink();

src/v5/default.rs

@@ -4,7 +4,7 @@ use ntex::service::{Service, ServiceFactory};
 use ntex::util::Ready;
 
 use super::control::{ControlMessage, ControlResult};
-use super::publish::{Publish, PublishAck};
+use super::publish::{PublishMessage, PublishResult};
 use super::Session;
 
 /// Default publish service
@@ -20,8 +20,8 @@ impl<St, Err> Default for DefaultPublishService<St, Err> {
 
 impl<St, Err> ServiceFactory for DefaultPublishService<St, Err> {
     type Config = Session<St>;
-    type Request = Publish;
-    type Response = PublishAck;
+    type Request = PublishMessage;
+    type Response = PublishResult;
     type Error = Err;
     type Service = DefaultPublishService<St, Err>;
     type InitError = Err;
@@ -33,16 +33,16 @@ impl<St, Err> ServiceFactory for DefaultPublishService<St, Err> {
 }
 
 impl<St, Err> Service for DefaultPublishService<St, Err> {
-    type Request = Publish;
-    type Response = PublishAck;
+    type Request = PublishMessage;
+    type Response = PublishResult;
     type Error = Err;
     type Future = Ready<Self::Response, Self::Error>;
 
     fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }
 
-    fn call(&self, req: Publish) -> Self::Future {
+    fn call(&self, req: PublishMessage) -> Self::Future {
         log::warn!("Publish service is disabled");
         Ready::Ok(req.ack())
     }