Issue #75 Validator MemoryChannelRepository (#91)

elpiel · web-flow · commit 21287ed0c3f1 · 2019-07-08T15:21:05.000+03:00
* validator - persistence - channel - memory:
- add module
- add MemoryChannelRepository struct
- impl `all()`

* validator - persistence - channel - api - no need for clone

* validator - persistence - channel - export Api &amp; Memory repositories

* domain - channel - SpecValidators - add `new()`

* validator - Cargo.toml - add `fixtures` feature for `domain`

* [+test] validator - persistence - channel - memory - add tests &amp; fix typos

* domain - channel - fixtures - create a ValidatorsOption to clarify &amp; clean `get_channel_spec`

* refactor to use the `ValidatorsOption` for `get_channel_spec`

* domain - validator - ValidatorId impl `AsRef&lt;str&gt;` &amp; `Display`

* validator - channel - ChannelRepository - use `ValidatorId` for `all()`
diff --git a/adapter/src/sanity.rs b/adapter/src/sanity.rs
@@ -105,7 +105,7 @@ impl error::Error for SanityError {
 mod test {
     use time::Duration;
 
-    use domain::channel::fixtures::get_channel_spec;
+    use domain::channel::fixtures::{get_channel_spec, ValidatorsOption};
     use domain::fixtures::{get_channel, get_validator};
 
     use crate::adapter::ConfigBuilder;
@@ -250,7 +250,7 @@ mod test {
             get_validator("my leader", Some(10.into())),
             get_validator("my follower", Some(15.into())),
         ];
-        let spec = get_channel_spec("channel_1", Some(validators.into()));
+        let spec = get_channel_spec(ValidatorsOption::SpecValidators(validators.into()));
         let channel = get_channel("channel_1", &None, Some(spec));
 
         // as identity use the leader, otherwise we won't pass the AdapterNotIncluded check
diff --git a/domain/src/channel.rs b/domain/src/channel.rs
@@ -30,7 +30,7 @@ impl fmt::Display for ChannelId {
     /// let channel_hex_string = format!("{}", channel_id);
     /// assert_eq!("0x061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088", &channel_hex_string);
     /// ```
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let hex_string = SerHex::<StrictPfx>::into_hex(&self.bytes).unwrap();
         write!(f, "{}", hex_string)
     }
@@ -183,6 +183,10 @@ impl<'a> SpecValidator<'a> {
 pub struct SpecValidators([ValidatorDesc; 2]);
 
 impl SpecValidators {
+    pub fn new(leader: ValidatorDesc, follower: ValidatorDesc) -> Self {
+        Self([leader, follower])
+    }
+
     pub fn leader(&self) -> &ValidatorDesc {
         &self.0[0]
     }
diff --git a/domain/src/channel_fixtures.rs b/domain/src/channel_fixtures.rs
@@ -7,7 +7,7 @@ use crate::fixtures::{get_targeting_tags, get_validator};
 use crate::test_util;
 use crate::BigNum;
 
-use super::{Channel, ChannelId, ChannelSpec, SpecValidators};
+use super::{Channel, ChannelId, ChannelSpec, SpecValidators, ValidatorDesc};
 
 /// It will get the length of channel_id bytes and will fill enough bytes in front
 /// If > 32 bytes &str is passed it will `panic!`
@@ -38,7 +38,11 @@ pub fn get_channel(
     });
     let creator = <Faker as Name>::name();
     let deposit_asset = get_asset();
-    let spec = spec.unwrap_or_else(|| get_channel_spec(id, None));
+    let spec = spec.unwrap_or_else(|| {
+        get_channel_spec(ValidatorsOption::Generate {
+            validators_prefix: id,
+        })
+    });
 
     Channel {
         id: channel_id,
@@ -63,15 +67,27 @@ pub fn get_channels(count: usize, valid_until_ge: Option<DateTime<Utc>>) -> Vec<
         .collect()
 }
 
-pub fn get_channel_spec(prefix: &str, validators_option: Option<SpecValidators>) -> ChannelSpec {
+pub enum ValidatorsOption<'a> {
+    Pair {
+        leader: ValidatorDesc,
+        follower: ValidatorDesc,
+    },
+    SpecValidators(SpecValidators),
+    Generate {
+        validators_prefix: &'a str,
+    },
+}
+
+pub fn get_channel_spec(validators_option: ValidatorsOption<'_>) -> ChannelSpec {
     use crate::EventSubmission;
     use test_util::take_one;
 
     let validators = match validators_option {
-        Some(validators) => validators,
-        None => [
-            get_validator(&format!("{} leader", prefix), None),
-            get_validator(&format!("{} follower", prefix), None),
+        ValidatorsOption::Pair { leader, follower } => [leader, follower].into(),
+        ValidatorsOption::SpecValidators(spec_validators) => spec_validators,
+        ValidatorsOption::Generate { validators_prefix } => [
+            get_validator(&format!("{} leader", validators_prefix), None),
+            get_validator(&format!("{} follower", validators_prefix), None),
         ]
         .into(),
     };
diff --git a/domain/src/validator.rs b/domain/src/validator.rs
@@ -1,4 +1,5 @@
 use std::convert::TryFrom;
+use std::fmt;
 
 use serde::{Deserialize, Serialize};
 
@@ -21,6 +22,18 @@ impl TryFrom<&str> for ValidatorId {
     }
 }
 
+impl AsRef<str> for ValidatorId {
+    fn as_ref(&self) -> &str {
+        self.0.as_str()
+    }
+}
+
+impl fmt::Display for ValidatorId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct ValidatorDesc {
@@ -38,12 +51,15 @@ pub mod fixtures {
 
     use super::ValidatorDesc;
 
-    pub fn get_validator(validator_id: &str, fee: Option<BigNum>) -> ValidatorDesc {
+    pub fn get_validator<V: AsRef<str>>(validator_id: V, fee: Option<BigNum>) -> ValidatorDesc {
         let fee = fee.unwrap_or_else(|| BigNum::from(<Faker as Number>::between(1, 13)));
-        let url = format!("http://{}-validator-url.com/validator", validator_id);
+        let url = format!(
+            "http://{}-validator-url.com/validator",
+            validator_id.as_ref()
+        );
 
         ValidatorDesc {
-            id: validator_id.to_string(),
+            id: validator_id.as_ref().to_string(),
             url,
             fee,
         }
diff --git a/sentry/src/infrastructure/persistence/channel/memory_test.rs b/sentry/src/infrastructure/persistence/channel/memory_test.rs
@@ -134,8 +134,8 @@ fn listing_channels_can_handles_validator_filtration_and_keeps_valid_until_filtr
             get_validator("validator-2", None),
         ]
         .into();
-        let channel_2_spec = get_channel_spec("channel 2", Some(validators.clone()));
-        let channel_5_spec = get_channel_spec("channel 5", Some(validators));
+        let channel_2_spec = get_channel_spec(ValidatorsOption::SpecValidators(validators.clone()));
+        let channel_5_spec = get_channel_spec(ValidatorsOption::SpecValidators(validators));
 
         let channels = [
             get_channel("channel 1", &None, None),
diff --git a/validator/Cargo.toml b/validator/Cargo.toml
@@ -10,7 +10,7 @@ path = "src/lib.rs"
 
 [dependencies]
 # Domain
-domain = { version = "0.1", path = "../domain", features = ["repositories"] }
+domain = { version = "0.1", path = "../domain", features = ["repositories", "fixtures"] }
 adapter = { version = "0.1", path = "../adapter", features = ["dummy-adapter"] }
 memory-repository = { version = "0.1", path = "../memory-repository" }
 chrono = { version = "0.4", features = ["serde"] }
diff --git a/validator/src/application/worker.rs b/validator/src/application/worker.rs
@@ -2,14 +2,15 @@ pub use self::infinite::InfiniteWorker;
 pub use self::single::TickWorker;
 
 pub mod single {
+    use std::convert::TryFrom;
     use std::sync::Arc;
     use std::time::Duration;
 
     use futures::compat::Future01CompatExt;
     use futures::future::{FutureExt, TryFutureExt};
     use tokio::util::FutureExt as TokioFutureExt;
 
-    use domain::{Channel, SpecValidator};
+    use domain::{Channel, SpecValidator, ValidatorId};
 
     use crate::application::validator::{Follower, Leader};
     use crate::domain::{ChannelRepository, Validator, Worker, WorkerFuture};
@@ -28,7 +29,10 @@ pub mod single {
     /// Single tick worker
     impl TickWorker {
         pub async fn tick(self) -> Result<(), ()> {
-            let all_channels = await!(self.channel_repository.all(&self.identity));
+            // @TODO: Update once we figure out if ValidatorId can fail from a &str
+            let validator_id = ValidatorId::try_from(self.identity.as_str())
+                .expect("ValidatorId doesn't have a failing case right now");
+            let all_channels = await!(self.channel_repository.all(&validator_id));
 
             match all_channels {
                 Ok(channels) => {
diff --git a/validator/src/domain/channel.rs b/validator/src/domain/channel.rs
@@ -1,6 +1,6 @@
-use domain::{Channel, RepositoryFuture};
+use domain::{Channel, RepositoryFuture, ValidatorId};
 
 pub trait ChannelRepository: Send + Sync {
     /// Returns list of all channels, based on the passed validator identity
-    fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>>;
+    fn all(&self, identity: &ValidatorId) -> RepositoryFuture<Vec<Channel>>;
 }
diff --git a/validator/src/infrastructure/persistence/channel.rs b/validator/src/infrastructure/persistence/channel.rs
@@ -1 +1,5 @@
+pub use self::api::ApiChannelRepository;
+pub use self::memory::MemoryChannelRepository;
+
 pub mod api;
+pub mod memory;
diff --git a/validator/src/infrastructure/persistence/channel/api.rs b/validator/src/infrastructure/persistence/channel/api.rs
@@ -1,22 +1,21 @@
 use futures::{FutureExt, TryFutureExt};
 
-use domain::{Channel, RepositoryFuture};
+use domain::{Channel, RepositoryFuture, ValidatorId};
 
 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 sentry: SentryApi,
 }
 
 impl ChannelRepository for ApiChannelRepository {
-    fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>> {
+    fn all(&self, identity: &ValidatorId) -> RepositoryFuture<Vec<Channel>> {
         self.sentry
             .clone()
-            .all_channels(Some(identity.to_string()))
+            .all_channels(Some(identity))
             // @TODO: Error handling
             .map_err(|_error| ApiPersistenceError::Reading.into())
             .boxed()
diff --git a/validator/src/infrastructure/persistence/channel/memory.rs b/validator/src/infrastructure/persistence/channel/memory.rs
@@ -0,0 +1,82 @@
+use std::sync::Arc;
+
+use futures::future::{ready, FutureExt};
+
+use domain::{Channel, ChannelId, RepositoryFuture, SpecValidator, ValidatorId};
+use memory_repository::MemoryRepository;
+
+use crate::domain::channel::ChannelRepository;
+
+// @TODO: make pub(crate)
+pub struct MemoryChannelRepository {
+    inner: MemoryRepository<Channel, ChannelId>,
+}
+
+impl MemoryChannelRepository {
+    pub fn new(initial_channels: &[Channel]) -> Self {
+        let cmp = Arc::new(|channel: &Channel, channel_id: &ChannelId| &channel.id == channel_id);
+
+        Self {
+            inner: MemoryRepository::new(&initial_channels, cmp),
+        }
+    }
+}
+
+impl ChannelRepository for MemoryChannelRepository {
+    fn all(&self, identity: &ValidatorId) -> RepositoryFuture<Vec<Channel>> {
+        let list =
+            self.inner.list_all(
+                |channel| match channel.spec.validators.find(identity.as_ref()) {
+                    SpecValidator::Leader(_) | SpecValidator::Follower(_) => Some(channel.clone()),
+                    SpecValidator::None => None,
+                },
+            );
+
+        ready(list.map_err(Into::into)).boxed()
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::convert::TryFrom;
+
+    use domain::fixtures::{
+        get_channel, get_channel_id, get_channel_spec, get_validator, ValidatorsOption,
+    };
+    use domain::ValidatorId;
+
+    use super::*;
+
+    #[test]
+    fn find_all_channels_with_the_passed_identity_and_skips_the_rest() {
+        futures::executor::block_on(async {
+            let identity = ValidatorId::try_from("Lookup identity").unwrap();
+            let validator_1 = get_validator(&identity, None);
+            let validator_2 = get_validator("Second", None);
+            let channel_1_spec = get_channel_spec(ValidatorsOption::Pair {
+                leader: validator_1.clone(),
+                follower: validator_2.clone(),
+            });
+
+            // switch leader & follower
+            let channel_3_spec = get_channel_spec(ValidatorsOption::Pair {
+                leader: validator_2.clone(),
+                follower: validator_1.clone(),
+            });
+            let channels = vec![
+                get_channel("channel 1", &None, Some(channel_1_spec)),
+                get_channel("channel 2", &None, None),
+                get_channel("channel 3", &None, Some(channel_3_spec)),
+            ];
+
+            let repository = MemoryChannelRepository::new(&channels);
+
+            let result: Vec<Channel> =
+                await!(repository.all(&identity)).expect("Getting all channels failed");
+
+            assert_eq!(2, result.len());
+            assert_eq!(&get_channel_id("channel 1"), &result[0].id);
+            assert_eq!(&get_channel_id("channel 3"), &result[1].id);
+        })
+    }
+}
diff --git a/validator/src/infrastructure/sentry.rs b/validator/src/infrastructure/sentry.rs
@@ -1,4 +1,4 @@
-use domain::Channel;
+use domain::{Channel, ValidatorId};
 use futures::compat::Future01CompatExt;
 use futures::future::{ok, try_join_all, FutureExt, TryFutureExt};
 use futures::Future;
@@ -18,8 +18,9 @@ pub struct SentryApi {
 impl SentryApi {
     pub fn all_channels(
         &self,
-        validator: Option<String>,
+        validator: Option<&ValidatorId>,
     ) -> impl Future<Output = Result<Vec<Channel>, Error>> {
+        let validator = validator.cloned();
         // call Sentry and fetch first page, where validator = identity
         let first_page = self.clone().fetch_page(1, validator.clone());
 
@@ -59,7 +60,7 @@ impl SentryApi {
     async fn fetch_page(
         self,
         page: u64,
-        validator: Option<String>,
+        validator: Option<ValidatorId>,
     ) -> Result<ChannelAllResponse, reqwest::Error> {
         let mut query = vec![format!("page={}", page)];
 

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`		`-use domain::{Channel, RepositoryFuture};`
	`1`	`+use domain::{Channel, RepositoryFuture, ValidatorId};`
`2`	`2`
`3`	`3`	`pub trait ChannelRepository: Send + Sync {`
`4`	`4`	`/// Returns list of all channels, based on the passed validator identity`
`5`		`- fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>>;`
	`5`	`+ fn all(&self, identity: &ValidatorId) -> RepositoryFuture<Vec<Channel>>;`
`6`	`6`	`}`