Feature: Added unsubscribe method to MqttClient (#123)

src/client/connection.rs

@@ -456,6 +456,7 @@ impl From<Request> for Packet {
             Request::Ping => Packet::Pingreq,
             Request::Disconnect => Packet::Disconnect,
             Request::Subscribe(subscribe) => Packet::Subscribe(subscribe),
+            Request::Unsubscribe(unsubscribe) => Packet::Unsubscribe(unsubscribe),
             _ => unimplemented!(),
         }
     }

src/client/mod.rs

@@ -2,7 +2,7 @@ use crate::error::{ClientError, ConnectError};
 use crate::MqttOptions;
 use crossbeam_channel;
 use futures::{sync::mpsc, Future, Sink};
-use mqtt311::{PacketIdentifier, Publish, QoS, Subscribe, SubscribeTopic};
+use mqtt311::{PacketIdentifier, Publish, QoS, Subscribe, Unsubscribe, SubscribeTopic};
 use std::sync::Arc;
 
 pub mod connection;
@@ -26,6 +26,7 @@ pub enum Notification {
 pub enum Request {
     Publish(Publish),
     Subscribe(Subscribe),
+    Unsubscribe(Unsubscribe),
     PubAck(PacketIdentifier),
     PubRec(PacketIdentifier),
     PubRel(PacketIdentifier),
@@ -119,6 +120,20 @@ impl MqttClient {
         Ok(())
     }
 
+    pub fn unsubscribe<S>(&mut self, topic: S) -> Result<(), ClientError>
+        where
+            S: Into<String>,
+    {
+        let unsubscribe = Unsubscribe {
+            pkid: PacketIdentifier::zero(),
+            topics: vec![topic.into()],
+        };
+
+        let tx = &mut self.request_tx;
+        tx.send(Request::Unsubscribe(unsubscribe)).wait()?;
+        Ok(())
+    }
+
     pub fn pause(&mut self) -> Result<(), ClientError> {
         let tx = &mut self.command_tx;
         tx.send(Command::Pause).wait()?;

src/client/mqttstate.rs

@@ -91,6 +91,7 @@ impl MqttState {
             Packet::Pingresp => self.handle_incoming_pingresp(),
             Packet::Publish(publish) => self.handle_incoming_publish(publish.clone()),
             Packet::Suback(_pkid) => Ok((Notification::None, Request::None)),
+            Packet::Unsuback(_pkid) => Ok((Notification::None, Request::None)),
             Packet::Puback(pkid) => self.handle_incoming_puback(pkid),
             Packet::Pubrec(pkid) => self.handle_incoming_pubrec(pkid),
             Packet::Pubrel(pkid) => self.handle_incoming_pubrel(pkid),