Issue #66 Worker main loop (#67)

* validator - infra - validator - follower & leader add `derive(Clone)``

* validator - infra - worker - single & infinite tick workers

* validator - infra - validator - re-export `Leader` & `Follower`

* validator - domain - worker - add `Worker` trait & `WorkerFuture`

* validator - infra - worker - impl `Worker` and cleanup for `TickWorker` & `InfiniteWorker`

* validator - main - use `TickWorker` & `InfiniteWorker`

* validator - domain - Validator - `tick()` to method and add `Channel`

* validator - infra - Worker:
- add handle_channel and call the validators
- add identity to the worker
- Call the `ChannelRepository::all()` with the worker identity

* fix `clippy` warnings

Author: elpiel

validator/src/domain.rs

@@ -1,2 +1,5 @@
+pub use self::worker::{Worker, WorkerFuture};
+
 pub mod channel;
 pub mod validator;
+pub mod worker;

validator/src/domain/validator.rs

@@ -1,6 +1,9 @@
-use futures::Future;
 use std::pin::Pin;
 
+use futures::Future;
+
+use domain::Channel;
+
 pub type ValidatorFuture<T> = Pin<Box<dyn Future<Output = Result<T, ValidatorError>> + Send>>;
 
 #[derive(Debug)]
@@ -9,5 +12,5 @@ pub enum ValidatorError {
 }
 
 pub trait Validator {
-    fn tick() -> ValidatorFuture<()>;
+    fn tick(&self, channel: Channel) -> ValidatorFuture<()>;
 }

validator/src/domain/worker.rs

@@ -0,0 +1,8 @@
+use futures::Future;
+use std::pin::Pin;
+
+pub type WorkerFuture = Pin<Box<dyn Future<Output = Result<(), ()>> + Send>>;
+
+pub trait Worker {
+    fn run(&self) -> WorkerFuture;
+}

validator/src/infrastructure.rs

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

validator/src/infrastructure/validator.rs

@@ -1,2 +1,5 @@
+pub use self::follower::Follower;
+pub use self::leader::Leader;
+
 pub mod follower;
 pub mod leader;

validator/src/infrastructure/validator/follower.rs

@@ -1,9 +1,13 @@
+use domain::Channel;
+
 use crate::domain::validator::{Validator, ValidatorFuture};
+use futures::future::FutureExt;
 
+#[derive(Clone)]
 pub struct Follower {}
 
 impl Validator for Follower {
-    fn tick() -> ValidatorFuture<()> {
-        unimplemented!()
+    fn tick(&self, _channel: Channel) -> ValidatorFuture<()> {
+        futures::future::ok(()).boxed()
     }
 }

validator/src/infrastructure/validator/leader.rs

@@ -1,9 +1,13 @@
+use domain::Channel;
+
 use crate::domain::validator::{Validator, ValidatorFuture};
+use futures::future::FutureExt;
 
+#[derive(Clone)]
 pub struct Leader {}
 
 impl Validator for Leader {
-    fn tick() -> ValidatorFuture<()> {
-        unimplemented!()
+    fn tick(&self, _channel: Channel) -> ValidatorFuture<()> {
+        futures::future::ok(()).boxed()
     }
 }

validator/src/infrastructure/worker.rs

@@ -0,0 +1,106 @@
+pub use self::infinite::InfiniteWorker;
+pub use self::single::TickWorker;
+
+pub mod single {
+    use std::sync::Arc;
+
+    use futures::future::FutureExt;
+
+    use crate::domain::channel::ChannelRepository;
+    use crate::domain::validator::Validator;
+    use crate::domain::{Worker, WorkerFuture};
+    use crate::infrastructure::validator::follower::Follower;
+    use crate::infrastructure::validator::leader::Leader;
+    use domain::channel::SpecValidator;
+    use domain::Channel;
+
+    #[derive(Clone)]
+    pub struct TickWorker {
+        pub leader: Leader,
+        pub follower: Follower,
+        pub channel_repository: Arc<dyn ChannelRepository>,
+        pub identity: String,
+    }
+
+    /// Single tick worker
+    impl TickWorker {
+        pub async fn tick(self) -> Result<(), ()> {
+            let all_channels = await!(self.channel_repository.all(&self.identity));
+
+            match all_channels {
+                Ok(channels) => {
+                    for channel in channels {
+                        await!(self.clone().handle_channel(channel)).unwrap();
+                    }
+                }
+                Err(error) => eprintln!("Error occurred: {:#?}", error),
+            };
+
+            Ok(())
+        }
+
+        async fn handle_channel(self, channel: Channel) -> Result<(), ()> {
+            let channel_id = channel.id;
+
+            match &channel.spec.validators.find(&self.identity) {
+                SpecValidator::Leader(_) => {
+                    self.leader.tick(channel);
+                    eprintln!("Channel {} handled as __Leader__", channel_id.to_string());
+                }
+                SpecValidator::Follower(_) => {
+                    self.follower.tick(channel);
+                    eprintln!("Channel {} handled as __Follower__", channel_id.to_string());
+                }
+                SpecValidator::None => {
+                    eprintln!("Channel {} is not validated by us", channel_id.to_string());
+                }
+            };
+
+            Ok(())
+        }
+    }
+
+    impl Worker for TickWorker {
+        fn run(&self) -> WorkerFuture {
+            self.clone().tick().boxed()
+        }
+    }
+}
+pub mod infinite {
+    use std::ops::Add;
+    use std::time::{Duration, Instant};
+
+    use futures::compat::Future01CompatExt;
+    use futures::future::{join, FutureExt};
+    use tokio::timer::Delay;
+
+    use crate::domain::{Worker, WorkerFuture};
+    use crate::infrastructure::worker::TickWorker;
+
+    #[derive(Clone)]
+    pub struct InfiniteWorker {
+        pub tick_worker: TickWorker,
+    }
+
+    /// Infinite tick worker
+    impl InfiniteWorker {
+        pub async fn infinite(self) -> Result<(), ()> {
+            let handle = self.clone();
+            loop {
+                let future = handle.clone().tick_worker.tick();
+                //                let tick_future = Delay::new(Instant::now().add(CONFIG.ticks_wait_time));
+                let tick_future = Delay::new(Instant::now().add(Duration::from_millis(5000)));
+
+                let joined = join(future, tick_future.compat());
+
+                await!(joined);
+            }
+        }
+    }
+
+    impl Worker for InfiniteWorker {
+        fn run(&self) -> WorkerFuture {
+            self.clone().infinite().boxed()
+        }
+    }
+}

validator/src/main.rs

@@ -2,17 +2,17 @@
 #![deny(rust_2018_idioms)]
 #![deny(clippy::all)]
 
-use futures::compat::Future01CompatExt;
-use futures::future::{join, FutureExt, TryFutureExt};
+use futures::future::{FutureExt, TryFutureExt};
 use reqwest::r#async::Client;
 
 use lazy_static::lazy_static;
-use std::ops::Add;
-use std::time::{Duration, Instant};
-use tokio::timer::Delay;
-use validator::domain::channel::ChannelRepository;
+use std::sync::Arc;
+use std::time::Duration;
+use validator::domain::worker::Worker;
 use validator::infrastructure::persistence::channel::api::ApiChannelRepository;
 use validator::infrastructure::sentry::SentryApi;
+use validator::infrastructure::validator::{Follower, Leader};
+use validator::infrastructure::worker::{InfiniteWorker, TickWorker};
 
 lazy_static! {
     static ref CONFIG: Config = {
@@ -33,31 +33,25 @@ lazy_static! {
 }
 
 fn main() {
-    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"));
-
-                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 sentry = SentryApi {
+        client: Client::new(),
+        sentry_url: CONFIG.sentry_url.clone(),
+    };
 
-            let joined = join(future, tick_future.compat());
+    let channel_repository = Arc::new(ApiChannelRepository {
+        sentry: sentry.clone(),
+    });
 
-            await!(joined);
-        }
+    let tick_worker = TickWorker {
+        leader: Leader {},
+        follower: Follower {},
+        channel_repository,
+        identity: "0x2892f6C41E0718eeeDd49D98D648C789668cA67d".to_string(),
     };
 
-    tokio::run(worker.unit_error().boxed().compat());
+    let worker = InfiniteWorker { tick_worker };
+
+    tokio::run(worker.run().boxed().compat());
 }
 
 struct Config {