test(contract-test): completed contract testing for subscribe

Completed set of contract tests for subscription event engine.

refactor(subscribe-ee): refactored retry

Refactored retry to be coupled with cancellable request.

refactor(clippy): apply clippy suggestions

Author: parfeon

Cargo.toml

@@ -109,7 +109,7 @@ getrandom = { version = "0.2", optional = true }
 ciborium = { version = "0.2.1", default-features = false, optional = true }
 
 # subscribe
-tokio = { version = "1", optional = true, features = ["rt-multi-thread", "macros"] }
+tokio = { version = "1", optional = true, features = ["rt-multi-thread", "macros", "time"] }
 futures = { version = "0.3.28", optional = true }
 async-channel = { version = "1.8", optional = true }
 
@@ -121,7 +121,7 @@ getrandom = { version = "0.2", features = ["js"] }
 
 [dev-dependencies]
 async-trait = "0.1"
-tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
 wiremock = "0.5"
 env_logger = "0.10"
 cucumber = { version = "0.20.0", features = ["output-junit"] }

src/core/error.rs

@@ -4,7 +4,10 @@
 //!
 //! [`pubnub`]: ../index.html
 
-use crate::lib::{alloc::string::String, alloc::vec::Vec};
+use crate::{
+    core::TransportResponse,
+    lib::alloc::{boxed::Box, string::String, vec::Vec},
+};
 use snafu::Snafu;
 
 /// PubNub error type
@@ -37,7 +40,7 @@ pub enum PubNubError {
         details: String,
 
         /// Failed request HTTP status code.
-        status: u16,
+        response: Option<Box<TransportResponse>>,
     },
 
     /// this error is returned when the publication of the request fails
@@ -132,6 +135,9 @@ pub enum PubNubError {
 
         /// List of channel groups which is affected by error.
         affected_channel_groups: Option<Vec<String>>,
+
+        /// Raw service response.
+        response: Option<Box<TransportResponse>>,
     },
 }
 
@@ -140,8 +146,12 @@ impl PubNubError {
     ///
     /// This function used to inform about not initialized request parameters or
     /// validation failure.
-    #[cfg(any(feature = "publish", feature = "access"))]
-    pub(crate) fn general_api_error<S>(message: S, status: Option<u16>) -> Self
+    #[cfg(any(feature = "publish", feature = "access", feature = "subscribe"))]
+    pub(crate) fn general_api_error<S>(
+        message: S,
+        status: Option<u16>,
+        response: Option<Box<TransportResponse>>,
+    ) -> Self
     where
         S: Into<String>,
     {
@@ -151,6 +161,48 @@ impl PubNubError {
             service: None,
             affected_channels: None,
             affected_channel_groups: None,
+            response,
+        }
+    }
+
+    /// Retrieve attached service response.
+    #[cfg(any(feature = "publish", feature = "access", feature = "subscribe"))]
+    pub(crate) fn transport_response(&self) -> Option<Box<TransportResponse>> {
+        match self {
+            PubNubError::API { response, .. } | PubNubError::Transport { response, .. } => {
+                response.clone()
+            }
+            _ => None,
+        }
+    }
+
+    /// Attach service response.
+    ///
+    /// For better understanding some errors may provide additional information
+    /// right from service response.
+    #[cfg(any(feature = "publish", feature = "access", feature = "subscribe"))]
+    pub(crate) fn attach_response(self, service_response: TransportResponse) -> Self {
+        match &self {
+            PubNubError::API {
+                status,
+                message,
+                service,
+                affected_channels,
+                affected_channel_groups,
+                ..
+            } => PubNubError::API {
+                status: *status,
+                message: message.clone(),
+                service: service.clone(),
+                affected_channels: affected_channels.clone(),
+                affected_channel_groups: affected_channel_groups.clone(),
+                response: Some(Box::new(service_response)),
+            },
+            PubNubError::Transport { details, .. } => PubNubError::Transport {
+                details: details.clone(),
+                response: Some(Box::new(service_response)),
+            },
+            _ => self,
         }
     }
 }

src/core/error_response.rs

@@ -24,6 +24,7 @@ impl From<APIErrorBody> for PubNubError {
             service: value.service(),
             affected_channels: value.affected_channels(),
             affected_channel_groups: value.affected_channel_groups(),
+            response: None,
         }
     }
 }

src/core/event_engine/effect_dispatcher.rs

@@ -234,13 +234,18 @@ mod should {
     #[derive(Clone)]
     struct TestRuntime {}
 
+    #[async_trait::async_trait]
     impl Runtime for TestRuntime {
         fn spawn<R>(&self, _future: impl Future<Output = R> + Send + 'static)
         where
             R: Send + 'static,
         {
             // Do nothing.
         }
+
+        async fn sleep(self, _delay: u64) {
+            // Do nothing.
+        }
     }
 
     #[test]

src/core/event_engine/mod.rs

@@ -313,13 +313,18 @@ mod should {
     #[derive(Clone)]
     struct TestRuntime {}
 
+    #[async_trait::async_trait]
     impl Runtime for TestRuntime {
         fn spawn<R>(&self, future: impl Future<Output = R> + Send + 'static)
         where
             R: Send + 'static,
         {
             tokio::spawn(future);
         }
+
+        async fn sleep(self, delay: u64) {
+            tokio::time::sleep(tokio::time::Duration::from_secs(delay)).await
+        }
     }
 
     #[tokio::test]