refactor: Pending type to pending request and promise_tx

Author: Devdutt Shenoi

rumqttc/src/client.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 
 use crate::mqttbytes::{v4::*, QoS};
 use crate::{
-    valid_filter, valid_topic, AckPromise, ConnectionError, Event, EventLoop, MqttOptions,
+    valid_filter, valid_topic, AckPromise, ConnectionError, Event, EventLoop, MqttOptions, Pending,
     PromiseTx, Request,
 };
 
@@ -23,15 +23,15 @@ pub enum ClientError {
     TryRequest(Request),
 }
 
-impl From<SendError<(Request, Option<PromiseTx>)>> for ClientError {
-    fn from(e: SendError<(Request, Option<PromiseTx>)>) -> Self {
-        Self::Request(e.into_inner().0)
+impl From<SendError<Pending<Request>>> for ClientError {
+    fn from(e: SendError<Pending<Request>>) -> Self {
+        Self::Request(e.into_inner().request)
     }
 }
 
-impl From<TrySendError<(Request, Option<PromiseTx>)>> for ClientError {
-    fn from(e: TrySendError<(Request, Option<PromiseTx>)>) -> Self {
-        Self::TryRequest(e.into_inner().0)
+impl From<TrySendError<Pending<Request>>> for ClientError {
+    fn from(e: TrySendError<Pending<Request>>) -> Self {
+        Self::TryRequest(e.into_inner().request)
     }
 }
 
@@ -44,7 +44,7 @@ impl From<TrySendError<(Request, Option<PromiseTx>)>> for ClientError {
 /// from the broker, i.e. move ahead.
 #[derive(Clone, Debug)]
 pub struct AsyncClient {
-    request_tx: Sender<(Request, Option<PromiseTx>)>,
+    request_tx: Sender<Pending<Request>>,
 }
 
 impl AsyncClient {
@@ -64,7 +64,7 @@ impl AsyncClient {
     ///
     /// This is mostly useful for creating a test instance where you can
     /// listen on the corresponding receiver.
-    pub fn from_senders(request_tx: Sender<(Request, Option<PromiseTx>)>) -> AsyncClient {
+    pub fn from_senders(request_tx: Sender<Pending<Request>>) -> AsyncClient {
         AsyncClient { request_tx }
     }
 
@@ -84,12 +84,11 @@ impl AsyncClient {
         let topic = topic.into();
         let mut publish = Publish::new(&topic, qos, payload);
         publish.retain = retain;
-        let publish = Request::Publish(publish);
         if !valid_topic(&topic) {
-            return Err(ClientError::Request(publish));
+            return Err(ClientError::Request(publish.into()));
         }
         self.request_tx
-            .send_async((publish, Some(promise_tx)))
+            .send_async(Pending::new(publish.into(), promise_tx))
             .await?;
 
         Ok(promise)
@@ -111,11 +110,11 @@ impl AsyncClient {
         let topic = topic.into();
         let mut publish = Publish::new(&topic, qos, payload);
         publish.retain = retain;
-        let publish = Request::Publish(publish);
         if !valid_topic(&topic) {
-            return Err(ClientError::TryRequest(publish));
+            return Err(ClientError::TryRequest(publish.into()));
         }
-        self.request_tx.try_send((publish, Some(promise_tx)))?;
+        self.request_tx
+            .try_send(Pending::new(publish.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -125,7 +124,9 @@ impl AsyncClient {
         let ack = get_ack_req(publish);
 
         if let Some(ack) = ack {
-            self.request_tx.send_async((ack, None)).await?;
+            self.request_tx
+                .send_async(Pending::no_promises(ack))
+                .await?;
         }
 
         Ok(())
@@ -135,7 +136,7 @@ impl AsyncClient {
     pub fn try_ack(&self, publish: &Publish) -> Result<(), ClientError> {
         let ack = get_ack_req(publish);
         if let Some(ack) = ack {
-            self.request_tx.try_send((ack, None))?;
+            self.request_tx.try_send(Pending::no_promises(ack))?;
         }
 
         Ok(())
@@ -155,9 +156,8 @@ impl AsyncClient {
         let (promise_tx, promise) = PromiseTx::new();
         let mut publish = Publish::from_bytes(topic, qos, payload);
         publish.retain = retain;
-        let publish = Request::Publish(publish);
         self.request_tx
-            .send_async((publish, Some(promise_tx)))
+            .send_async(Pending::new(publish.into(), promise_tx))
             .await?;
 
         Ok(promise)
@@ -175,7 +175,7 @@ impl AsyncClient {
             return Err(ClientError::Request(subscribe.into()));
         }
         self.request_tx
-            .send_async((subscribe.into(), Some(promise_tx)))
+            .send_async(Pending::new(subscribe.into(), promise_tx))
             .await?;
 
         Ok(promise)
@@ -193,7 +193,7 @@ impl AsyncClient {
             return Err(ClientError::TryRequest(subscribe.into()));
         }
         self.request_tx
-            .try_send((subscribe.into(), Some(promise_tx)))?;
+            .try_send(Pending::new(subscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -209,7 +209,7 @@ impl AsyncClient {
             return Err(ClientError::Request(subscribe.into()));
         }
         self.request_tx
-            .send_async((subscribe.into(), Some(promise_tx)))
+            .send_async(Pending::new(subscribe.into(), promise_tx))
             .await?;
 
         Ok(promise)
@@ -226,7 +226,7 @@ impl AsyncClient {
             return Err(ClientError::TryRequest(subscribe.into()));
         }
         self.request_tx
-            .try_send((subscribe.into(), Some(promise_tx)))?;
+            .try_send(Pending::new(subscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -236,7 +236,7 @@ impl AsyncClient {
         let (promise_tx, promise) = PromiseTx::new();
         let unsubscribe = Unsubscribe::new(topic.into());
         self.request_tx
-            .send_async((unsubscribe.into(), Some(promise_tx)))
+            .send_async(Pending::new(unsubscribe.into(), promise_tx))
             .await?;
 
         Ok(promise)
@@ -247,23 +247,25 @@ impl AsyncClient {
         let (promise_tx, promise) = PromiseTx::new();
         let unsubscribe = Unsubscribe::new(topic.into());
         self.request_tx
-            .try_send((unsubscribe.into(), Some(promise_tx)))?;
+            .try_send(Pending::new(unsubscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
 
     /// Sends a MQTT disconnect to the `EventLoop`
     pub async fn disconnect(&self) -> Result<(), ClientError> {
         let request = Request::Disconnect(Disconnect);
-        self.request_tx.send_async((request, None)).await?;
+        self.request_tx
+            .send_async(Pending::no_promises(request))
+            .await?;
 
         Ok(())
     }
 
     /// Attempts to send a MQTT disconnect to the `EventLoop`
     pub fn try_disconnect(&self) -> Result<(), ClientError> {
         let request = Request::Disconnect(Disconnect);
-        self.request_tx.try_send((request, None))?;
+        self.request_tx.try_send(Pending::no_promises(request))?;
 
         Ok(())
     }
@@ -313,7 +315,7 @@ impl Client {
     ///
     /// This is mostly useful for creating a test instance where you can
     /// listen on the corresponding receiver.
-    pub fn from_sender(request_tx: Sender<(Request, Option<PromiseTx>)>) -> Client {
+    pub fn from_sender(request_tx: Sender<Pending<Request>>) -> Client {
         Client {
             client: AsyncClient::from_senders(request_tx),
         }
@@ -335,11 +337,13 @@ impl Client {
         let topic = topic.into();
         let mut publish = Publish::new(&topic, qos, payload);
         publish.retain = retain;
-        let publish = Request::Publish(publish);
+        let request = Request::Publish(publish);
         if !valid_topic(&topic) {
-            return Err(ClientError::Request(publish));
+            return Err(ClientError::Request(request));
         }
-        self.client.request_tx.send((publish, Some(promise_tx)))?;
+        self.client
+            .request_tx
+            .send(Pending::new(request, promise_tx))?;
 
         Ok(promise)
     }
@@ -363,7 +367,7 @@ impl Client {
         let ack = get_ack_req(publish);
 
         if let Some(ack) = ack {
-            self.client.request_tx.send((ack, None))?;
+            self.client.request_tx.send(Pending::no_promises(ack))?;
         }
 
         Ok(())
@@ -387,7 +391,7 @@ impl Client {
         }
         self.client
             .request_tx
-            .send((subscribe.into(), Some(promise_tx)))?;
+            .send(Pending::new(subscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -413,7 +417,7 @@ impl Client {
         }
         self.client
             .request_tx
-            .send((subscribe.into(), Some(promise_tx)))?;
+            .send(Pending::new(subscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -429,8 +433,9 @@ impl Client {
     pub fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<AckPromise, ClientError> {
         let (promise_tx, promise) = PromiseTx::new();
         let unsubscribe = Unsubscribe::new(topic.into());
-        let request = Request::Unsubscribe(unsubscribe);
-        self.client.request_tx.send((request, Some(promise_tx)))?;
+        self.client
+            .request_tx
+            .send(Pending::new(unsubscribe.into(), promise_tx))?;
 
         Ok(promise)
     }
@@ -444,7 +449,9 @@ impl Client {
     pub fn disconnect(&self) -> Result<AckPromise, ClientError> {
         let (promise_tx, promise) = PromiseTx::new();
         let request = Request::Disconnect(Disconnect);
-        self.client.request_tx.send((request, Some(promise_tx)))?;
+        self.client
+            .request_tx
+            .send(Pending::new(request, promise_tx))?;
 
         Ok(promise)
     }

rumqttc/src/eventloop.rs

@@ -1,5 +1,5 @@
 use crate::{framed::Network, Transport};
-use crate::{Incoming, MqttState, NetworkOptions, Packet, PromiseTx, Request, StateError};
+use crate::{Incoming, MqttState, NetworkOptions, Packet, Pending, Request, StateError};
 use crate::{MqttOptions, Outgoing};
 
 use crate::framed::AsyncReadWrite;
@@ -75,11 +75,11 @@ pub struct EventLoop {
     /// Current state of the connection
     pub state: MqttState,
     /// Request stream
-    requests_rx: Receiver<(Request, Option<PromiseTx>)>,
+    requests_rx: Receiver<Pending<Request>>,
     /// Requests handle to send requests
-    pub(crate) requests_tx: Sender<(Request, Option<PromiseTx>)>,
+    pub(crate) requests_tx: Sender<Pending<Request>>,
     /// Pending packets from last session
-    pub pending: VecDeque<(Request, Option<PromiseTx>)>,
+    pub pending: VecDeque<Pending<Request>>,
     /// Network connection to the broker
     pub network: Option<Network>,
     /// Keep alive time
@@ -132,7 +132,7 @@ impl EventLoop {
         // drain requests from channel which weren't yet received
         let mut requests_in_channel: Vec<_> = self.requests_rx.drain().collect();
 
-        requests_in_channel.retain(|(request, _)| {
+        requests_in_channel.retain(|Pending { request, .. }| {
             match request {
                 Request::PubAck(_) => false, // Wait for publish retransmission, else the broker could be confused by an unexpected ack
                 _ => true,
@@ -184,10 +184,10 @@ impl EventLoop {
     /// Select on network and requests and generate keepalive pings when necessary
     async fn select(&mut self) -> Result<Event, ConnectionError> {
         let network = self.network.as_mut().unwrap();
+        let network_timeout = Duration::from_secs(self.network_options.connection_timeout());
         // let await_acks = self.state.await_acks;
         let inflight_full = self.state.inflight >= self.mqtt_options.inflight;
         let collision = self.state.collision.is_some();
-        let network_timeout = Duration::from_secs(self.network_options.connection_timeout());
 
         // Read buffered events from previous polls before calling a new poll
         if let Some(event) = self.state.events.pop_front() {
@@ -241,8 +241,8 @@ impl EventLoop {
                 &self.requests_rx,
                 self.mqtt_options.pending_throttle
             ), if !self.pending.is_empty() || (!inflight_full && !collision) => match o {
-                Ok((request, tx)) => {
-                    if let Some(outgoing) = self.state.handle_outgoing_packet(request, tx)? {
+                Ok(pending) => {
+                    if let Some(outgoing) = self.state.handle_outgoing_packet(pending)? {
                         network.write(outgoing).await?;
                     }
                     match time::timeout(network_timeout, network.flush()).await {
@@ -257,21 +257,37 @@ impl EventLoop {
             // simple. We can change this behavior in future if necessary (to prevent extra pings)
             _ = self.keepalive_timeout.as_mut().unwrap_or(&mut no_sleep),
                 if self.keepalive_timeout.is_some() && !self.mqtt_options.keep_alive.is_zero() => {
-                let timeout = self.keepalive_timeout.as_mut().unwrap();
-                timeout.as_mut().reset(Instant::now() + self.mqtt_options.keep_alive);
-
-                if let Some(outgoing) = self.state.handle_outgoing_packet(Request::PingReq(PingReq), None)? {
-                    network.write(outgoing).await?;
-                }
-                match time::timeout(network_timeout, network.flush()).await {
-                    Ok(inner) => inner?,
-                    Err(_)=> return Err(ConnectionError::FlushTimeout),
-                };
+                self.handle_timeout().await?;
                 Ok(self.state.events.pop_front().unwrap())
             }
         }
     }
 
+    async fn handle_timeout(&mut self) -> Result<(), ConnectionError> {
+        let network = self.network.as_mut().unwrap();
+        let network_timeout = Duration::from_secs(self.network_options.connection_timeout());
+
+        // Set to next timeout instant
+        self.keepalive_timeout
+            .as_mut()
+            .unwrap()
+            .as_mut()
+            .reset(Instant::now() + self.mqtt_options.keep_alive);
+
+        if let Some(outgoing) = self
+            .state
+            .handle_outgoing_packet(Pending::no_promises(Request::PingReq(PingReq)))?
+        {
+            network.write(outgoing).await?;
+        }
+        match time::timeout(network_timeout, network.flush()).await {
+            Ok(inner) => inner?,
+            Err(_) => return Err(ConnectionError::FlushTimeout),
+        };
+
+        Ok(())
+    }
+
     pub fn network_options(&self) -> NetworkOptions {
         self.network_options.clone()
     }
@@ -282,10 +298,10 @@ impl EventLoop {
     }
 
     async fn next_request(
-        pending: &mut VecDeque<(Request, Option<PromiseTx>)>,
-        rx: &Receiver<(Request, Option<PromiseTx>)>,
+        pending: &mut VecDeque<Pending<Request>>,
+        rx: &Receiver<Pending<Request>>,
         pending_throttle: Duration,
-    ) -> Result<(Request, Option<PromiseTx>), ConnectionError> {
+    ) -> Result<Pending<Request>, ConnectionError> {
         if !pending.is_empty() {
             time::sleep(pending_throttle).await;
             // We must call .pop_front() AFTER sleep() otherwise we would have