Issue 60 make sentry url configurable (#65)

* domain - repository - remove Stream future

* validator - infrastructure - sentry - add sentry API struct

* validator - persistence - channel - api - use `SentryApi`

* cleanup

* Config

* validator - Cargo.toml - add `dotenv` for now

* validator - load `sentry_url` from `.env` file for now

* validator - main - POC for the validator ticks

* run `rustfmt`

Author: elpiel

.env.dist

@@ -1 +1,3 @@
-CHANNEL_LIST_LIMIT=200
+SENTRY_CHANNEL_LIST_LIMIT=200
+
+VALIDATOR_TICKS_WAIT_TIME=500

Cargo.lock

@@ -55,7 +55,7 @@ impl error::Error for DomainError {
 pub mod repository {
     use std::pin::Pin;
 
-    use futures::{Future, Stream};
+    use futures::Future;
 
     pub trait IOError: std::error::Error + Send {}
 
@@ -69,6 +69,4 @@ pub mod repository {
     }
 
     pub type RepositoryFuture<T> = Pin<Box<dyn Future<Output = Result<T, RepositoryError>> + Send>>;
-
-    pub type RepositoryStream<T> = Pin<Box<dyn Stream<Item = Result<T, RepositoryError>> + Send>>;
 }

domain/src/lib.rs

@@ -82,19 +82,19 @@ impl TryFrom<Vars> for Config {
     fn try_from(mut vars: Vars) -> Result<Self, Self::Error> {
         let limit = vars
             .find_map(|(key, value)| {
-                if key == "CHANNEL_LIST_LIMIT" {
+                if key == "SENTRY_CHANNEL_LIST_LIMIT" {
                     Some(value)
                 } else {
                     None
                 }
             })
             .ok_or(DomainError::InvalidArgument(
-                "CHANNEL_LIST_LIMIT evn. variable was not passed".to_string(),
+                "SENTRY_CHANNEL_LIST_LIMIT evn. variable was not passed".to_string(),
             ))
             .and_then(|value| {
                 value.parse::<u32>().map_err(|_| {
                     DomainError::InvalidArgument(
-                        "CHANNEL_LIST_LIMIT is not a u32 value".to_string(),
+                        "SENTRY_CHANNEL_LIST_LIMIT is not a u32 value".to_string(),
                     )
                 })
             });

sentry/src/main.rs

@@ -18,6 +18,9 @@ futures_legacy = { version = "0.1", package = "futures" }
 tokio = { version = "=0.1.19" }
 # API client
 reqwest = "0.9.18"
+# Configuration
+lazy_static = "1.3"
+dotenv = "0.14"
 # (De)Serialization
 serde = { version = "^1.0", features = ['derive'] }
 serde_json = "1.0"

validator/Cargo.toml

@@ -1,2 +1,3 @@
 pub mod persistence;
+pub mod sentry;
 pub mod validator;

validator/src/infrastructure.rs

@@ -1,86 +1,24 @@
-use std::iter::once;
-use std::sync::Arc;
-
-use futures::compat::Future01CompatExt;
-use futures::future::{ok, try_join_all};
 use futures::{FutureExt, TryFutureExt};
-use futures_legacy::Future as LegacyFuture;
-use reqwest::r#async::{Client, Response};
-use serde::Deserialize;
 
 use domain::{Channel, RepositoryFuture};
 
 use crate::domain::channel::ChannelRepository;
 use crate::infrastructure::persistence::api::ApiPersistenceError;
+use crate::infrastructure::sentry::SentryApi;
 
 #[derive(Clone)]
+// @TODO: make pub(crate)
 pub struct ApiChannelRepository {
-    pub client: Client,
-}
-
-impl ApiChannelRepository {
-    fn fetch_page(&self, page: u64, identity: &str) -> RepositoryFuture<ChannelAllResponse> {
-        self.client
-            // call Sentry and fetch first page, where validator = identity
-            .get(
-                format!(
-                    "http://localhost:8005/channel/list?validator={}&page={}",
-                    identity, page
-                )
-                .as_str(),
-            )
-            .send()
-            .and_then(|mut res: Response| res.json::<ChannelAllResponse>())
-            // @TODO: Error handling
-            .map_err(|_error| ApiPersistenceError::Reading.into())
-            .compat()
-            .boxed()
-    }
+    pub sentry: SentryApi,
 }
 
 impl ChannelRepository for ApiChannelRepository {
     fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>> {
-        let identity = Arc::new(identity.to_string());
-        let handle = self.clone();
-
-        let first_page = handle.fetch_page(1, &identity.clone());
-
-        // call Sentry again and concat all the Channels in Future
-        // fetching them until no more Channels are returned
-        first_page
-            .and_then(move |response| {
-                let first_page_future = ok(response.channels).boxed();
-
-                if response.total_pages < 2 {
-                    first_page_future
-                } else {
-                    let identity = identity.clone();
-                    let futures = (2..=response.total_pages)
-                        .map(|page| {
-                            handle
-                                .fetch_page(page, &identity)
-                                .map(|response_result| {
-                                    response_result.and_then(|response| Ok(response.channels))
-                                })
-                                .boxed()
-                        })
-                        .chain(once(first_page_future));
-
-                    try_join_all(futures)
-                        .map(|result_all| {
-                            result_all
-                                .and_then(|all| Ok(all.into_iter().flatten().collect::<Vec<_>>()))
-                        })
-                        .boxed()
-                }
-            })
+        self.sentry
+            .clone()
+            .all_channels(Some(identity.to_string()))
+            // @TODO: Error handling
+            .map_err(|_error| ApiPersistenceError::Reading.into())
             .boxed()
     }
 }
-
-#[derive(Deserialize, Debug)]
-#[serde(rename_all = "camelCase")]
-struct ChannelAllResponse {
-    pub channels: Vec<Channel>,
-    pub total_pages: u64,
-}

validator/src/infrastructure/persistence/channel/api.rs

@@ -0,0 +1,85 @@
+use domain::Channel;
+use futures::compat::Future01CompatExt;
+use futures::future::{ok, try_join_all, FutureExt, TryFutureExt};
+use futures::Future;
+use futures_legacy::Future as LegacyFuture;
+use reqwest::r#async::{Client, Response};
+use reqwest::Error;
+use serde::Deserialize;
+use std::iter::once;
+
+#[derive(Clone)]
+// @TODO: make pub(crate)
+pub struct SentryApi {
+    pub sentry_url: String,
+    pub client: Client,
+}
+
+impl SentryApi {
+    pub fn all_channels(
+        &self,
+        validator: Option<String>,
+    ) -> impl Future<Output = Result<Vec<Channel>, Error>> {
+        let first_page = self.clone().fetch_page(1, validator.clone());
+
+        // call Sentry again and concat all the Channels in Future
+        // fetching them until no more Channels are returned
+        let handle = self.clone();
+        first_page
+            .and_then(move |response| {
+                let first_page_future = ok(response.channels).boxed();
+
+                if response.total_pages < 2 {
+                    first_page_future
+                } else {
+                    let futures = (2..=response.total_pages)
+                        .map(|page| {
+                            handle
+                                .clone()
+                                .fetch_page(page, validator.clone())
+                                .map(|response_result| {
+                                    response_result.and_then(|response| Ok(response.channels))
+                                })
+                                .boxed()
+                        })
+                        .chain(once(first_page_future));
+
+                    try_join_all(futures)
+                        .map(|result_all| {
+                            result_all
+                                .and_then(|all| Ok(all.into_iter().flatten().collect::<Vec<_>>()))
+                        })
+                        .boxed()
+                }
+            })
+            .boxed()
+    }
+
+    async fn fetch_page(
+        self,
+        page: u64,
+        validator: Option<String>,
+    ) -> Result<ChannelAllResponse, reqwest::Error> {
+        let mut query = vec![format!("page={}", page)];
+
+        if let Some(validator) = validator {
+            query.push(format!("validator={}", validator));
+        }
+
+        let future = self
+            .client
+            // call Sentry and fetch first page, where validator = identity
+            .get(format!("{}/channel/list?{}", self.sentry_url, query.join("&")).as_str())
+            .send()
+            .and_then(|mut res: Response| res.json::<ChannelAllResponse>());
+
+        await!(future.compat())
+    }
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+struct ChannelAllResponse {
+    pub channels: Vec<Channel>,
+    pub total_pages: u64,
+}

validator/src/infrastructure/sentry.rs

@@ -1,24 +1,66 @@
 #![feature(async_await, await_macro)]
 #![deny(rust_2018_idioms)]
 #![deny(clippy::all)]
+
+use futures::compat::Future01CompatExt;
+use futures::future::{join, FutureExt, TryFutureExt};
 use reqwest::r#async::Client;
 
-use futures::future::{FutureExt, TryFutureExt};
+use lazy_static::lazy_static;
+use std::ops::Add;
+use std::time::{Duration, Instant};
+use tokio::timer::Delay;
 use validator::domain::channel::ChannelRepository;
+use validator::infrastructure::persistence::channel::api::ApiChannelRepository;
+use validator::infrastructure::sentry::SentryApi;
+
+lazy_static! {
+    static ref CONFIG: Config = {
+        dotenv::dotenv().ok();
+
+        let ticks_wait_time = std::env::var("VALIDATOR_TICKS_WAIT_TIME")
+            .unwrap()
+            .parse()
+            .unwrap();
+        Config {
+            ticks_wait_time: Duration::from_millis(ticks_wait_time),
+            sentry_url: std::env::var("VALIDATOR_SENTRY_URL")
+                .unwrap()
+                .parse()
+                .unwrap(),
+        }
+    };
+}
 
 fn main() {
-    let future = async {
-        let repo = validator::infrastructure::persistence::channel::api::ApiChannelRepository {
-            client: Client::new(),
-        };
+    let worker = async {
+        loop {
+            let future = async {
+                let sentry = SentryApi {
+                    client: Client::new(),
+                    sentry_url: CONFIG.sentry_url.clone(),
+                };
+                let repo = ApiChannelRepository { sentry };
 
-        let all_channels = await!(repo.all("0x2892f6C41E0718eeeDd49D98D648C789668cA67d"));
+                let all_channels = await!(repo.all("0x2892f6C41E0718eeeDd49D98D648C789668cA67d"));
 
-        match all_channels {
-            Ok(channel) => println!("{:#?}", channel),
-            Err(error) => eprintln!("Error occurred: {:#?}", error),
-        };
+                match all_channels {
+                    Ok(channel) => println!("{:#?}", channel),
+                    Err(error) => eprintln!("Error occurred: {:#?}", error),
+                };
+            };
+            let tick_future = Delay::new(Instant::now().add(CONFIG.ticks_wait_time));
+
+            let joined = join(future, tick_future.compat());
+
+            await!(joined);
+        }
     };
 
-    tokio::run(future.unit_error().boxed().compat());
+    tokio::run(worker.unit_error().boxed().compat());
+}
+
+struct Config {
+    pub ticks_wait_time: Duration,
+    pub sentry_url: String,
 }