fix: don't panic on connection closure (#833)

rumqttc/CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 * `size()` method on `Packet` calculates size once serialized.
 * `read()` and `write()` methods on `Packet`.
+* `ConnectionAborted` variant on `StateError` type to denote abrupt end to a connection
 
 ### Changed
 

rumqttc/src/framed.rs

@@ -38,8 +38,9 @@ impl Network {
     pub async fn read(&mut self) -> Result<Incoming, StateError> {
         match self.framed.next().await {
             Some(Ok(packet)) => Ok(packet),
-            Some(Err(mqttbytes::Error::InsufficientBytes(_))) | None => unreachable!(),
+            Some(Err(mqttbytes::Error::InsufficientBytes(_))) => unreachable!(),
             Some(Err(e)) => Err(StateError::Deserialization(e)),
+            None => Err(StateError::ConnectionAborted),
         }
     }
 
@@ -61,8 +62,9 @@ impl Network {
                         break;
                     }
                 }
-                Some(Err(mqttbytes::Error::InsufficientBytes(_))) | None => unreachable!(),
+                Some(Err(mqttbytes::Error::InsufficientBytes(_))) => unreachable!(),
                 Some(Err(e)) => return Err(StateError::Deserialization(e)),
+                None => return Err(StateError::ConnectionAborted),
             }
             // do not wait for subsequent reads
             match self.framed.next().now_or_never() {

rumqttc/src/state.rs

@@ -29,6 +29,8 @@ pub enum StateError {
     EmptySubscription,
     #[error("Mqtt serialization/deserialization error: {0}")]
     Deserialization(#[from] mqttbytes::Error),
+    #[error("Connection closed by peer abruptly")]
+    ConnectionAborted,
 }
 
 /// State of the mqtt connection.

rumqttc/src/v5/framed.rs

@@ -40,8 +40,9 @@ impl Network {
     pub async fn read(&mut self) -> Result<Incoming, StateError> {
         match self.framed.next().await {
             Some(Ok(packet)) => Ok(packet),
-            Some(Err(mqttbytes::Error::InsufficientBytes(_))) | None => unreachable!(),
+            Some(Err(mqttbytes::Error::InsufficientBytes(_))) => unreachable!(),
             Some(Err(e)) => Err(StateError::Deserialization(e)),
+            None => Err(StateError::ConnectionAborted),
         }
     }
 
@@ -63,8 +64,9 @@ impl Network {
                         break;
                     }
                 }
-                Some(Err(mqttbytes::Error::InsufficientBytes(_))) | None => unreachable!(),
+                Some(Err(mqttbytes::Error::InsufficientBytes(_))) => unreachable!(),
                 Some(Err(e)) => return Err(StateError::Deserialization(e)),
+                None => return Err(StateError::ConnectionAborted),
             }
             // do not wait for subsequent reads
             match self.framed.next().now_or_never() {

rumqttc/src/v5/state.rs

@@ -64,6 +64,8 @@ pub enum StateError {
     PubCompFail { reason: PubCompReason },
     #[error("Connection failed with reason '{reason:?}' ")]
     ConnFail { reason: ConnectReturnCode },
+    #[error("Connection closed by peer abruptly")]
+    ConnectionAborted
 }
 
 impl From<mqttbytes::Error> for StateError {