worker - SentryApi - move collect_channels

Author: elpiel

primitives/src/address.rs

@@ -111,11 +111,11 @@ impl TryFrom<&str> for Address {
 /// # Example
 /// ```
 /// use once_cell::sync::Lazy;
-/// use crate::Address;
+/// use primitives::Address;
 ///
 /// static ADDRESS_0: Lazy<Address> = Lazy::new(|| b"0x80690751969B234697e9059e04ed72195c3507fa".try_into().unwrap());
 ///
-/// println!("Address: {}", ADDRESS_0);
+/// println!("Address: {}", *ADDRESS_0);
 /// ```
 impl TryFrom<&'static [u8; 42]> for Address {
     type Error = Error;

validator_worker/src/channel.rs

@@ -1,15 +1,10 @@
 use crate::{
     error::{Error, TickError},
-    follower, leader,
-    sentry_interface::{campaigns::all_campaigns, Validator, Validators},
-    SentryApi,
-};
-use primitives::{adapter::Adapter, config::Config, util::ApiUrl, Channel, ChannelId};
-use slog::{info, Logger};
-use std::{
-    collections::{hash_map::Entry, HashSet},
-    time::Duration,
+    follower, leader, SentryApi,
 };
+use primitives::{adapter::Adapter, config::Config, Channel, ChannelId};
+use slog::info;
+use std::time::Duration;
 use tokio::time::timeout;
 
 pub async fn channel_tick<A: Adapter + 'static>(
@@ -93,67 +88,3 @@ pub async fn channel_tick<A: Adapter + 'static>(
         }
     }
 }
-
-/// Fetches all `Campaign`s from Sentry and builds the `Channel`s to be processed
-/// along side all the `Validator`s' url & auth token
-// TODO: Move to [`SentryApi`]
-pub async fn collect_channels<A: Adapter + 'static>(
-    adapter: &A,
-    sentry_url: &ApiUrl,
-    config: &Config,
-    _logger: &Logger,
-) -> Result<(HashSet<Channel>, Validators), reqwest::Error> {
-    let for_whoami = adapter.whoami();
-
-    let all_campaigns_timeout = Duration::from_millis(config.all_campaigns_timeout as u64);
-    let client = reqwest::Client::builder()
-        .timeout(all_campaigns_timeout)
-        .build()?;
-
-    let whoami_validator = Validator {
-        url: sentry_url.clone(),
-        token: adapter
-            .get_auth(&for_whoami)
-            .expect("Should get WhoAmI auth"),
-    };
-    let campaigns = all_campaigns(client, &whoami_validator, Some(for_whoami)).await?;
-    let channels = campaigns
-        .iter()
-        .map(|campaign| campaign.channel)
-        .collect::<HashSet<_>>();
-
-    let validators = campaigns
-        .into_iter()
-        .fold(Validators::new(), |mut acc, campaign| {
-            for validator_desc in campaign.validators.iter() {
-                // if Validator is already there, we can just skip it
-                // remember, the campaigns are ordered by `created DESC`
-                // so we will always get the latest Validator url first
-                match acc.entry(validator_desc.id) {
-                    Entry::Occupied(_) => continue,
-                    Entry::Vacant(entry) => {
-                        // try to parse the url of the Validator Desc
-                        let validator_url = validator_desc.url.parse::<ApiUrl>();
-                        // and also try to find the Auth token in the config
-
-                        // if there was an error with any of the operations, skip this `ValidatorDesc`
-                        let auth_token = adapter.get_auth(&validator_desc.id);
-
-                        // only if `ApiUrl` parsing is `Ok` & Auth Token is found in the `Adapter`
-                        if let (Ok(url), Ok(auth_token)) = (validator_url, auth_token) {
-                            // add an entry for propagation
-                            entry.insert(Validator {
-                                url,
-                                token: auth_token,
-                            });
-                        }
-                        // otherwise it will try to do the same things on the next encounter of this `ValidatorId`
-                    }
-                }
-            }
-
-            acc
-        });
-
-    Ok((channels, validators))
-}

validator_worker/src/main.rs

@@ -107,7 +107,7 @@ fn main() -> Result<(), Box<dyn Error>> {
             };
             let mut adapter = DummyAdapter::init(options, &config);
             // unlock adapter
-            adapter.unlock().expect("failed to Unlock Ethereum adapter");
+            adapter.unlock().expect("failed to Unlock Dummy adapter");
 
             AdapterTypes::DummyAdapter(Box::new(adapter))
         }

validator_worker/src/sentry_interface.rs

@@ -1,4 +1,7 @@
-use std::{collections::HashMap, time::Duration};
+use std::{
+    collections::{hash_map::Entry, HashMap, HashSet},
+    time::Duration,
+};
 
 use futures::future::{join_all, try_join_all, TryFutureExt};
 use reqwest::{Client, Method};
@@ -14,7 +17,7 @@ use primitives::{
     spender::Spender,
     util::ApiUrl,
     validator::MessageTypes,
-    Address, ChannelId, Config, ValidatorId,
+    Address, Channel, ChannelId, Config, ValidatorId,
 };
 use thiserror::Error;
 
@@ -248,6 +251,57 @@ impl<A: Adapter + 'static, P> SentryApi<A, P> {
             .map_err(Error::Request)
             .await?)
     }
+
+    /// Fetches all `Campaign`s from the _Who am I_ Sentry.
+    /// It builds the `Channel`s to be processed alongside all the `Validator`s' url & auth token.
+    pub async fn collect_channels(&self) -> Result<(HashSet<Channel>, Validators), Error> {
+        let all_campaigns_timeout = Duration::from_millis(self.config.all_campaigns_timeout as u64);
+        let client = reqwest::Client::builder()
+            .timeout(all_campaigns_timeout)
+            .build()?;
+
+        let campaigns =
+            campaigns::all_campaigns(client, &self.whoami, Some(self.adapter.whoami())).await?;
+        let channels = campaigns
+            .iter()
+            .map(|campaign| campaign.channel)
+            .collect::<HashSet<_>>();
+
+        let validators = campaigns
+            .into_iter()
+            .fold(Validators::new(), |mut acc, campaign| {
+                for validator_desc in campaign.validators.iter() {
+                    // if Validator is already there, we can just skip it
+                    // remember, the campaigns are ordered by `created DESC`
+                    // so we will always get the latest Validator url first
+                    match acc.entry(validator_desc.id) {
+                        Entry::Occupied(_) => continue,
+                        Entry::Vacant(entry) => {
+                            // try to parse the url of the Validator Desc
+                            let validator_url = validator_desc.url.parse::<ApiUrl>();
+                            // and also try to find the Auth token in the config
+
+                            // if there was an error with any of the operations, skip this `ValidatorDesc`
+                            let auth_token = self.adapter.get_auth(&validator_desc.id);
+
+                            // only if `ApiUrl` parsing is `Ok` & Auth Token is found in the `Adapter`
+                            if let (Ok(url), Ok(auth_token)) = (validator_url, auth_token) {
+                                // add an entry for propagation
+                                entry.insert(Validator {
+                                    url,
+                                    token: auth_token,
+                                });
+                            }
+                            // otherwise it will try to do the same things on the next encounter of this `ValidatorId`
+                        }
+                    }
+                }
+
+                acc
+            });
+
+        Ok((channels, validators))
+    }
 }
 
 impl<A: Adapter + 'static> SentryApi<A> {

validator_worker/src/worker.rs

@@ -1,7 +1,4 @@
-use crate::{
-    channel::{channel_tick, collect_channels},
-    SentryApi,
-};
+use crate::{channel::channel_tick, SentryApi};
 use primitives::{adapter::Adapter, Config};
 use slog::{error, info, Logger};
 use std::{error::Error, time::Duration};
@@ -60,14 +57,8 @@ impl<A: Adapter + 'static> Worker<A> {
 
     pub async fn all_channels_tick(&self) {
         let logger = &self.logger;
-        let (channels, validators) = match collect_channels(
-            &self.adapter,
-            &self.sentry.whoami.url,
-            &self.config,
-            logger,
-        )
-        .await
-        {
+
+        let (channels, validators) = match self.sentry.collect_channels().await {
             Ok(res) => res,
             Err(err) => {
                 error!(logger, "Error collecting all channels for tick"; "collect_channels" => ?err, "main" => "all_channels_tick");