1 files changed, 26 insertions, 0 deletions
diff --git a/crates/common/mqtt_channel/src/errors.rs b/crates/common/mqtt_channel/src/errors.rs
index 6804dd94..505d1cb9 100644
--- a/crates/common/mqtt_channel/src/errors.rs
+++ b/crates/common/mqtt_channel/src/errors.rs
@@ -7,12 +7,22 @@ pub enum MqttError {
     #[error("Invalid topic filter: {pattern:?}")]
     InvalidFilter { pattern: String },
 
+    #[error("Invalid session: a session name must be provided")]
+    InvalidSessionConfig,
+
     #[error("MQTT client error: {0}")]
     ClientError(#[from] rumqttc::ClientError),
 
     #[error("MQTT connection error: {0}")]
     ConnectionError(#[from] rumqttc::ConnectionError),
 
+    #[error("MQTT connection rejected: {0:?}")]
+    ConnectionRejected(rumqttc::ConnectReturnCode),
+
+    #[error("MQTT subscription failure")]
+    // The MQTT specs are mysterious on the possible cause of such a failure
+    SubscriptionFailure,
+
     #[error("Invalid UTF8 payload: {from}: {input_excerpt}...")]
     InvalidUtf8Payload {
         input_excerpt: String,
@@ -31,6 +41,22 @@ pub enum MqttError {
 }
 
 impl MqttError {
+    pub fn maybe_connection_error(ack: &rumqttc::ConnAck) -> Option<MqttError> {
+        match ack.code {
+            rumqttc::ConnectReturnCode::Success => None,
+            err => Some(MqttError::ConnectionRejected(err)),
+        }
+    }
+
+    pub fn maybe_subscription_error(ack: &rumqttc::SubAck) -> Option<MqttError> {
+        for code in ack.return_codes.iter() {
+            if let rumqttc::SubscribeReasonCode::Failure = code {
+                return Some(MqttError::SubscriptionFailure);
+            }
+        }
+        None
+    }
+
     pub fn new_invalid_utf8_payload(bytes: &[u8], from: std::str::Utf8Error) -> MqttError {
         const EXCERPT_LEN: usize = 80;
         let index = from.valid_up_to();