Retry requests

Cargo.toml

@@ -16,15 +16,21 @@ base64 = "0.13"
 nanoserde = "0.1.26"
 urlencoding = "2.0.0-alpha.1"
 async-trait = "0.1.50"
+async-recursion = "0.3.2"
 oneshot = "0.1.2"
 log = "0.4.14"
 isahc = "1.4.0"
 qws = { version = "0.7.9", features = ["ssl"] }
 chrono = "0.4.19"
+url = "1.7.0"
+rand = "0.8.4"
+tokio-timer = "0.3.0-alpha.6"
 
 #[cfg(feature = "test")]
 futures = "0.3.15"
 
 [dev-dependencies]
 simple_logger = "1.11.0"
 cassette = "0.2.3"
+hyper = { version = "0.14.10", features = ["server", "http1", "tcp"]}
+tokio = { version = "1.8.1", features = ["macros"] }

src/client_adapter.rs

@@ -17,8 +17,13 @@ use async_trait::async_trait;
 use nanoserde::DeJson;
 use std::error::Error;
 
+pub trait ClientAdapterError: Error {
+   fn is_server_error(&self) -> bool;
+   fn is_client_error(&self) -> bool;
+}
+
 #[async_trait]
 pub trait ClientAdapter {
-    type Error: Error;
+    type Error: ClientAdapterError;
     async fn send<T: DeJson + Send>(&self, request: RestRequest<T>) -> Result<T, Self::Error>;
 }

src/default_client.rs

@@ -46,7 +46,7 @@ use crate::api::{
 };
 use crate::api_gen::{ApiSession, ApiWriteStorageObjectsRequest};
 use crate::client::Client;
-use crate::client_adapter::ClientAdapter;
+use crate::client_adapter::{ClientAdapter, ClientAdapterError};
 use crate::config::{DEFAULT_HOST, DEFAULT_PORT, DEFAULT_SERVER_KEY, DEFAULT_SERVER_PASSWORD};
 use crate::http_adapter::RestHttpAdapter;
 use crate::session::Session;
@@ -55,11 +55,19 @@ use nanoserde::DeJson;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::{Debug, Display, Formatter};
-
-pub struct DefaultClient<A: ClientAdapter> {
+use crate::retry::{RetryHistory, backoff, RetryConfiguration, Retry, DefaultDelay, Delay};
+use rand::{Rng, SeedableRng};
+use rand::rngs::StdRng;
+use std::sync::{Mutex, Arc};
+use async_recursion::async_recursion;
+use log::debug;
+
+pub struct DefaultClient<A: ClientAdapter, D: Delay = DefaultDelay> {
     adapter: A,
     server_key: String,
     server_password: String,
+    retry_configuration: Arc<Mutex<RetryConfiguration<StdRng, D>>>,
+    rng: Arc<Mutex<StdRng>>,
 }
 
 impl<A: ClientAdapter + Clone> Clone for DefaultClient<A> {
@@ -68,6 +76,8 @@ impl<A: ClientAdapter + Clone> Clone for DefaultClient<A> {
             adapter: self.adapter.clone(),
             server_key: self.server_key.clone(),
             server_password: self.server_password.clone(),
+            retry_configuration: self.retry_configuration.clone(),
+            rng: self.rng.clone(),
         }
     }
 }
@@ -96,15 +106,78 @@ impl DefaultClient<RestHttpAdapter> {
     }
 }
 
-impl<A: ClientAdapter + Send + Sync> DefaultClient<A> {
-    pub fn new(adapter: A, server_key: &str, server_password: &str) -> DefaultClient<A> {
+impl<A: ClientAdapter + Send + Sync> DefaultClient<A, DefaultDelay> {
+    pub fn new(adapter: A,
+               server_key: &str,
+               server_password: &str,
+    ) -> DefaultClient<A> {
+        let seed = [1, 0, 0, 0, 23, 0, 0, 0, 200, 1, 0, 0, 210, 30, 0, 0,
+            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
+        DefaultClient {
+            adapter,
+            server_key: server_key.to_owned(),
+            server_password: server_password.to_owned(),
+            rng: Arc::new(Mutex::new(StdRng::from_seed(seed))),
+            retry_configuration: Arc::new(Mutex::new(RetryConfiguration::new())),
+        }
+    }
+}
+
+
+impl<A: ClientAdapter + Send + Sync, D: Delay + Send> DefaultClient<A, D> {
+    pub fn new_with_configuration(adapter: A,
+               server_key: &str,
+               server_password: &str,
+        retry_configuration: RetryConfiguration<StdRng, D>
+    ) -> DefaultClient<A, D> {
+        let seed = [1,0,0,0, 23,0,0,0, 200,1,0,0, 210,30,0,0,
+            0,0,0,0, 0,0,0,0, 0,0,0,0, 0,0,0,0];
         DefaultClient {
             adapter,
             server_key: server_key.to_owned(),
             server_password: server_password.to_owned(),
+            rng: Arc::new(Mutex::new(StdRng::from_seed(seed))),
+            retry_configuration: Arc::new(Mutex::new(retry_configuration)),
         }
     }
 
+    pub async fn send_with_retry<T: DeJson + Send + Clone>(&self, request: RestRequest<T>) -> Result<T, DefaultClientError<A>> {
+       let history = RetryHistory::new(self.retry_configuration.clone());
+        self._send_with_retry(request, history, self.rng.clone()).await
+    }
+
+    #[async_recursion]
+    pub async fn _send_with_retry<T: DeJson + Send + Clone, R: Rng + Send>(&self, request: RestRequest<T>, retry_history: RetryHistory<R, D>, rng: Arc<Mutex<R>>) -> Result<T, DefaultClientError<A>>
+    {
+        {
+            let fut = self.send(request.clone());
+            let result = fut.await;
+            let retry = match result {
+                Ok(result) => return Ok(result),
+                Err(DefaultClientError::HttpAdapterError(ref err)) if err.is_client_error() || err.is_server_error() => {
+                    true
+                },
+                Err(err) => return Err(err)
+            };
+
+            if !retry {
+                return result;
+            }
+
+            let len = retry_history.retries.lock().expect("Failed to lock mutex").len();
+            let max_attempts = retry_history.retry_configuration.lock().expect("Failed to lock mutex").max_attempts;
+            if len >= max_attempts {
+                debug!("Exceeded {} retries.", max_attempts);
+                return result
+            }
+        }
+
+        debug!("Retrying request {}.", request.urlpath);
+
+        let new_history = backoff(retry_history, rng.clone()).await;
+        self._send_with_retry(request, new_history, rng).await
+    }
+
     #[inline]
     async fn send<T: DeJson + Send>(
         &self,
@@ -169,7 +242,7 @@ impl<A: ClientAdapter> Display for DefaultClientError<A> {
 impl<A: ClientAdapter> Error for DefaultClientError<A> {}
 
 #[async_trait]
-impl<A: ClientAdapter + Sync + Send> Client for DefaultClient<A> {
+impl<A: ClientAdapter + Sync + Send, D: Delay + Send> Client for DefaultClient<A, D> {
     type Error = DefaultClientError<A>;
 
     /// Add friends by id or username.
@@ -337,7 +410,7 @@ impl<A: ClientAdapter + Sync + Send> Client for DefaultClient<A> {
             username,
         );
 
-        self.send(request)
+        self.send_with_retry(request)
             .await
             .map(DefaultClient::<A>::map_session)
     }

src/http_adapter.rs

@@ -17,7 +17,7 @@ use std::error::Error;
 use std::fmt::{Display, Formatter};
 
 use crate::api;
-use crate::client_adapter::ClientAdapter;
+use crate::client_adapter::{ClientAdapter, ClientAdapterError};
 use async_trait::async_trait;
 use isahc::prelude::*;
 use nanoserde::{DeJson, DeJsonErr};
@@ -33,6 +33,22 @@ pub enum RestHttpError {
     OtherError(String),
 }
 
+impl ClientAdapterError for RestHttpError {
+    fn is_server_error(&self) -> bool {
+        match self {
+            RestHttpError::ServerError(_, _) => true,
+            _ => false
+        }
+    }
+
+    fn is_client_error(&self) -> bool {
+        match self {
+            RestHttpError::ClientError(_, _) => true,
+            _ => false
+        }
+    }
+}
+
 impl Display for RestHttpError {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         std::fmt::Debug::fmt(self, f)

src/lib.rs

@@ -244,8 +244,11 @@ pub mod socket;
 pub mod socket_adapter;
 #[cfg(feature = "test")]
 pub mod test_helpers;
+#[cfg(feature = "test")]
+pub mod mock_adapter;
 pub mod web_socket;
 pub mod web_socket_adapter;
+pub mod retry;
 
 pub use client::Client;
 pub use default_client::DefaultClient;

src/mock_adapter.rs

@@ -0,0 +1,41 @@
+use crate::client_adapter::{ClientAdapter, ClientAdapterError};
+use crate::api::RestRequest;
+use nanoserde::DeJson;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+use async_trait::async_trait;
+
+#[derive(Debug)]
+pub struct MockAdapterError {
+}
+
+impl Display for MockAdapterError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Debug::fmt(self, f)
+    }
+}
+
+impl Error for MockAdapterError {
+}
+
+impl ClientAdapterError for MockAdapterError {
+    fn is_server_error(&self) -> bool {
+        return true
+    }
+
+    fn is_client_error(&self) -> bool {
+        return false
+    }
+}
+
+pub struct MockAdapter {
+}
+
+#[async_trait]
+impl ClientAdapter for MockAdapter {
+    type Error = MockAdapterError;
+
+    async fn send<T: DeJson + Send>(&self, _request: RestRequest<T>) -> Result<T, Self::Error> {
+        return Err(MockAdapterError {})
+    }
+}