Use ClientOptions to create Pipeline

Author: heaths

sdk/core/src/client_options.rs

@@ -10,17 +10,17 @@ extern crate serde_derive;
 mod macros;
 
 mod bytes_stream;
-pub mod client_options;
 mod constants;
 mod context;
 mod errors;
 pub mod headers;
 mod http_client;
 pub mod incompletevector;
 mod models;
+mod options;
 pub mod parsing;
 pub mod pipeline;
-pub mod policies;
+mod policies;
 pub mod prelude;
 mod request;
 mod request_options;
@@ -36,13 +36,14 @@ use std::fmt::Debug;
 use uuid::Uuid;
 
 pub use bytes_stream::*;
-pub use client_options::ClientOptions;
 pub use constants::*;
 pub use context::Context;
 pub use errors::*;
 pub use headers::AddAsHeader;
 pub use http_client::{new_http_client, to_json, HttpClient};
 pub use models::*;
+pub use options::*;
+pub use policies::{Policy, PolicyResult};
 pub use request::*;
 pub use response::*;
 pub use seekable_stream::*;

sdk/core/src/lib.rs

@@ -15,7 +15,7 @@
 /// ```
 /// struct MyStruct<'a> { foo: Option<&'a str> };
 /// impl <'a> MyStruct<'a> {
-///     fn with_foo(self, foo: &'a str) -> Self {
+///     fn foo(self, foo: &'a str) -> Self {
 ///         Self {
 ///             foo: Some(foo),
 ///             ..self
@@ -26,6 +26,7 @@
 #[macro_export]
 macro_rules! setters {
     (@single $name:ident : $typ:ty => $transform:expr) => {
+        // TODO: Declare with_$name when https://github.com/rust-lang/rust/issues/29599 is fixed.
         pub fn $name<T: ::std::convert::Into<$typ>>(self, $name: T) -> Self {
             let $name: $typ = $name.into();
             Self  {
@@ -170,6 +171,28 @@ mod test {
     create_enum!(Colors, (Black, "Black"), (White, "White"), (Red, "Red"));
     create_enum!(ColorsMonochrome, (Black, "Black"), (White, "White"));
 
+    struct Options {
+        a: Option<String>,
+        b: u32,
+    }
+
+    #[allow(dead_code)]
+    impl Options {
+        setters! {
+            a: String => Some(a),
+            b: u32 => b,
+        }
+    }
+
+    impl Default for Options {
+        fn default() -> Self {
+            Options {
+                a: None,
+                b: 1,
+            }
+        }
+    }
+
     #[test]
     fn test_color_parse_1() {
         let color = "Black".parse::<Colors>().unwrap();
@@ -187,4 +210,13 @@ mod test {
     fn test_color_parse_err_1() {
         "Red".parse::<ColorsMonochrome>().unwrap();
     }
+
+    #[test]
+    fn test_setters() {
+        let options = Options::default()
+            .a("test".to_owned());
+
+        assert_eq!(Some("test".to_owned()), options.a);
+        assert_eq!(1, options.b);
+    }
 }

sdk/core/src/macros.rs

@@ -0,0 +1,153 @@
+use crate::policies::{ExponentialRetryPolicy, FixedRetryPolicy, NoRetryPolicy, Policy};
+use crate::{new_http_client, HttpClient};
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Client options allow customization of policies, retry options, and more.
+#[derive(Clone, Debug, Default)]
+pub struct ClientOptions {
+    // TODO: Expose transport override.
+    /// Policies called per call.
+    pub(crate) per_call_policies: Vec<Arc<dyn Policy>>,
+
+    /// Policies called per retry.
+    pub(crate) per_retry_policies: Vec<Arc<dyn Policy>>,
+
+    /// Retry options.
+    pub(crate) retry: RetryOptions,
+
+    /// Telemetry options.
+    pub(crate) telemetry: TelemetryOptions,
+
+    /// Transport options.
+    pub(crate) transport: TransportOptions,
+}
+
+impl ClientOptions {
+    setters! {
+        per_call_policies: Vec<Arc<dyn Policy>> => per_call_policies,
+        per_retry_policies: Vec<Arc<dyn Policy>> => per_retry_policies,
+        retry: RetryOptions => retry,
+        telemetry: TelemetryOptions => telemetry,
+        transport: TransportOptions => transport,
+    }
+}
+
+/// The algorithm to apply when calculating the delay between retry attempts.
+#[derive(Clone, Debug, PartialEq)]
+pub enum RetryMode {
+    /// Retry attempts will delay based on a back-off strategy,
+    /// where each attempt will increase the duration that it waits before retrying.
+    ///
+    /// This is the default.
+    Exponential,
+
+    /// Retry attempts happen at fixed intervals; each delay is a consistent duration.
+    Fixed,
+
+    /// Do not retry attempts.
+    None,
+}
+
+impl Default for RetryMode {
+    fn default() -> Self {
+        RetryMode::Exponential
+    }
+}
+
+/// The set of options that can be specified to influence how retry attempts are made,
+/// and a failure is eligible to be retried.
+#[derive(Clone, Debug)]
+pub struct RetryOptions {
+    /// The algorithm to use for calculating retry delays.
+    pub(crate) mode: RetryMode,
+
+    /// The delay between retry attempts for a fixed algorithm
+    /// or the delay on which to base calculations for a back-off-based approach.
+    ///
+    /// The default is 800 milliseconds.
+    pub(crate) delay: Duration,
+
+    /// The maximum number of retry attempts before giving up.
+    ///
+    /// The default is 3.
+    pub(crate) max_retries: u32,
+
+    /// The maximum permissible delay between retry attempts.
+    ///
+    /// The default is 1 minute.
+    pub(crate) max_delay: Duration,
+}
+
+impl RetryOptions {
+    setters! {
+        mode: RetryMode => mode,
+        delay: Duration => delay,
+        max_retries: u32 => max_retries,
+        max_delay: Duration => max_delay,
+    }
+}
+
+impl Default for RetryOptions {
+    fn default() -> Self {
+        RetryOptions {
+            mode: RetryMode::default(),
+            delay: Duration::from_millis(800),
+            max_retries: 3,
+            max_delay: Duration::from_secs(60),
+        }
+    }
+}
+
+impl RetryOptions {
+    pub(crate) fn to_policy(&self) -> Arc<dyn Policy> {
+        match self.mode {
+            RetryMode::Exponential => Arc::new(ExponentialRetryPolicy::new(
+                self.delay,
+                self.max_retries,
+                self.max_delay,
+            )),
+            RetryMode::Fixed => Arc::new(FixedRetryPolicy::new(
+                self.delay,
+                self.max_retries,
+                self.max_delay,
+            )),
+            RetryMode::None => Arc::new(NoRetryPolicy::default()),
+        }
+    }
+}
+
+/// Telemetry options.
+#[derive(Clone, Debug, Default)]
+pub struct TelemetryOptions {
+    /// Optional application ID to telemeter.
+    pub(crate) application_id: Option<String>,
+}
+
+impl TelemetryOptions {
+    setters! {
+        application_id: String => Some(application_id),
+    }
+}
+
+/// Transport options.
+#[derive(Clone, Debug)]
+pub struct TransportOptions {
+    /// The HTTP client implementation to use for requests.
+    pub(crate) http_client: Arc<dyn HttpClient>,
+}
+
+impl TransportOptions {
+    pub fn new(http_client: Arc<dyn HttpClient>) -> Self {
+        Self { http_client }
+    }
+}
+
+impl Default for TransportOptions {
+    /// Creates an instance of the ```TransportOptions``` using the enabled HTTP client.
+    fn default() -> Self {
+        TransportOptions {
+            http_client: new_http_client(),
+        }
+    }
+}

sdk/core/src/options.rs

@@ -1,4 +1,4 @@
-use crate::policies::{Policy, PolicyResult, TelemetryPolicy};
+use crate::policies::{Policy, PolicyResult, TelemetryPolicy, TransportPolicy};
 use crate::{ClientOptions, Context, Request, Response};
 use std::sync::Arc;
 
@@ -37,9 +37,7 @@ impl Pipeline {
         crate_version: Option<&'static str>,
         options: &ClientOptions,
         per_call_policies: Vec<Arc<dyn Policy>>,
-        retry: Arc<dyn Policy>,
         per_retry_policies: Vec<Arc<dyn Policy>>,
-        transport_policy: Arc<dyn Policy>,
     ) -> Self {
         let mut pipeline: Vec<Arc<dyn Policy>> = Vec::with_capacity(
             options.per_call_policies.len()
@@ -51,15 +49,21 @@ impl Pipeline {
 
         pipeline.extend_from_slice(&per_call_policies);
         pipeline.extend_from_slice(&options.per_call_policies);
+
         pipeline.push(Arc::new(TelemetryPolicy::new(
             crate_name,
             crate_version,
             &options.telemetry,
         )));
-        pipeline.push(retry);
+
+        let retry_policy = options.retry.to_policy();
+        pipeline.push(retry_policy);
+
         pipeline.extend_from_slice(&per_retry_policies);
         pipeline.extend_from_slice(&options.per_retry_policies);
-        pipeline.push(transport_policy);
+
+        let transport_policy = TransportPolicy::new(&options.transport);
+        pipeline.push(Arc::new(transport_policy));
 
         Self { pipeline }
     }

sdk/core/src/pipeline.rs

@@ -4,9 +4,9 @@ use chrono::{DateTime, Local};
 use std::sync::Arc;
 use std::time::Duration;
 
-/// Retry policy with exponential backoff.
+/// Retry policy with exponential back-off.
 ///
-/// Retry policy with exponential backoff (with an added random delay up to 256 ms). Each retry
+/// Retry policy with exponential back-off (with an added random delay up to 256 ms). Each retry
 /// will happen at least after an exponential wait time. So if x is the first retry wait, the
 /// second will be x*2, the third x*4 and so on. The policy will retry until the maximum number of
 /// retries have been reached or the maximum allowed delay has passed (whichever comes first). The
@@ -18,17 +18,15 @@ pub struct ExponentialRetryPolicy {
     max_delay: Duration,
 }
 
-impl Default for ExponentialRetryPolicy {
-    fn default() -> Self {
-        Self {
-            delay: Duration::from_secs(3),
-            max_retries: 3,
-            max_delay: Duration::from_secs(30),
+impl ExponentialRetryPolicy {
+    pub(crate) fn new(delay: Duration, max_retries: u32, max_delay: Duration) -> Self {
+        ExponentialRetryPolicy {
+            delay: delay,
+            max_retries: max_retries,
+            max_delay: max_delay,
         }
     }
-}
 
-impl ExponentialRetryPolicy {
     fn is_expired(
         &self,
         first_retry_time: &mut Option<DateTime<Local>>,

sdk/core/src/policies/retry_policies/exponential_retry.rs

@@ -4,30 +4,28 @@ use chrono::{DateTime, Local};
 use std::sync::Arc;
 use std::time::Duration;
 
-/// Retry policy with linear backoff.
+/// Retry policy with fixed back-off.
 ///
-/// Retry policy with linear backoff (with an added random delay up to 256 ms). Each retry will
+/// Retry policy with fixed back-off (with an added random delay up to 256 ms). Each retry will
 /// happen at least after the same, configured sleep time. The policy will retry until the maximum number of
 /// retries have been reached or the maximum allowed delay has passed (whichever comes first). The
 /// wait time is not precise.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct LinearRetryPolicy {
+pub struct FixedRetryPolicy {
     delay: Duration,
     max_retries: u32,
     max_delay: Duration,
 }
 
-impl Default for LinearRetryPolicy {
-    fn default() -> Self {
-        Self {
-            delay: Duration::from_secs(3),
-            max_retries: 3,
-            max_delay: Duration::from_secs(10),
+impl FixedRetryPolicy {
+    pub(crate) fn new(delay: Duration, max_retries: u32, max_delay: Duration) -> Self {
+        FixedRetryPolicy {
+            delay: delay,
+            max_retries: max_retries,
+            max_delay: max_delay,
         }
     }
-}
 
-impl LinearRetryPolicy {
     fn is_expired(
         &self,
         first_retry_time: &mut Option<DateTime<Local>>,
@@ -47,7 +45,7 @@ impl LinearRetryPolicy {
 }
 
 #[async_trait::async_trait]
-impl Policy for LinearRetryPolicy {
+impl Policy for FixedRetryPolicy {
     async fn send(
         &self,
         ctx: &mut Context,

sdk/core/src/policies/retry_policies/linear_retry.rs renamed to sdk/core/src/policies/retry_policies/fixed_retry.rs

@@ -1,7 +1,7 @@
 mod exponential_retry;
-mod linear_retry;
+mod fixed_retry;
 mod no_retry;
 
 pub use exponential_retry::*;
-pub use linear_retry::*;
+pub use fixed_retry::*;
 pub use no_retry::*;