Call control service on readiness error (#181)

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [4.2.0] - 2024-10-31
+
+* Call control service on readiness error
+
 ## [4.1.1] - 2024-10-15
 
 * Disconnect on error from service readiness check

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "4.1.1"
+version = "4.2.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "Client and Server framework for MQTT v5 and v3.1.1 protocols"
 documentation = "https://docs.rs/ntex-mqtt"
@@ -18,7 +18,7 @@ features = ["ntex/tokio"]
 ntex-io = "2"
 ntex-net = "2"
 ntex-util = "2"
-ntex-service = "3"
+ntex-service = "3.2.1"
 ntex-bytes = "0.1"
 ntex-codec = "0.6"
 ntex-router = "0.5"

src/error.rs

@@ -10,6 +10,9 @@ pub enum MqttError<E> {
     /// Publish handler service error
     #[error("Service error")]
     Service(E),
+    /// Publish service readiness error
+    #[error("Service readiness error")]
+    Readiness(Option<E>),
     /// Handshake error
     #[error("Mqtt handshake error: {}", _0)]
     Handshake(#[from] HandshakeError<E>),

src/v3/client/dispatcher.rs

@@ -69,8 +69,18 @@ where
     #[inline]
     async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         let (res1, res2) = join(ctx.ready(&self.publish), ctx.ready(&self.inner.control)).await;
-        res1.map_err(MqttError::Service)?;
-        res2
+        if let Err(e) = res1 {
+            if res2.is_err() {
+                Err(MqttError::Readiness(Some(e)))
+            } else {
+                match ctx.call_nowait(&self.inner.control, Control::error(e)).await {
+                    Ok(_) => Err(MqttError::Readiness(None)),
+                    Err(err) => Err(err),
+                }
+            }
+        } else {
+            res2
+        }
     }
 
     async fn shutdown(&self) {

src/v3/dispatcher.rs

@@ -128,8 +128,18 @@ where
     #[inline]
     async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         let (res1, res2) = join(ctx.ready(&self.publish), ctx.ready(&self.inner.control)).await;
-        res1.map_err(|e| MqttError::Service(e.into()))?;
-        res2
+        if let Err(e) = res1 {
+            if res2.is_err() {
+                Err(MqttError::Readiness(Some(e.into())))
+            } else {
+                match ctx.call_nowait(&self.inner.control, Control::error(e.into())).await {
+                    Ok(_) => Err(MqttError::Readiness(None)),
+                    Err(err) => Err(err),
+                }
+            }
+        } else {
+            res2
+        }
     }
 
     async fn shutdown(&self) {

src/v5/client/dispatcher.rs

@@ -95,8 +95,18 @@ where
     #[inline]
     async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         let (res1, res2) = join(ctx.ready(&self.publish), ctx.ready(&self.inner.control)).await;
-        res1.map_err(MqttError::Service)?;
-        res2
+        if let Err(e) = res1 {
+            if res2.is_err() {
+                Err(MqttError::Readiness(Some(e)))
+            } else {
+                match ctx.call_nowait(&self.inner.control, Control::error(e)).await {
+                    Ok(_) => Err(MqttError::Readiness(None)),
+                    Err(err) => Err(err),
+                }
+            }
+        } else {
+            res2
+        }
     }
 
     async fn shutdown(&self) {

src/v5/dispatcher.rs

@@ -136,8 +136,18 @@ where
     #[inline]
     async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         let (res1, res2) = join(ctx.ready(&self.publish), ctx.ready(&self.inner.control)).await;
-        res1.map_err(|e| MqttError::Service(e.into()))?;
-        res2
+        if let Err(e) = res1 {
+            if res2.is_err() {
+                Err(MqttError::Readiness(Some(e.into())))
+            } else {
+                match ctx.call_nowait(&self.inner.control, Control::error(e.into())).await {
+                    Ok(_) => Err(MqttError::Readiness(None)),
+                    Err(err) => Err(err),
+                }
+            }
+        } else {
+            res2
+        }
     }
 
     async fn shutdown(&self) {