Issue #93 Validator Message propagation (#98)

* validator - message_propagation - MessagePropagator

* [test] validator - message_propagation - MessagePropagator

* validator - application - HeartbeatSender - use MessagePropagator

Author: elpiel

validator/src/application.rs

@@ -1,3 +1,6 @@
 pub mod heartbeat;
+pub mod message_propagation;
 pub mod validator;
 pub mod worker;
+
+pub use self::message_propagation::MessagePropagator;

validator/src/application/heartbeat.rs

@@ -7,6 +7,7 @@ use adapter::{Adapter, AdapterError, BalanceRoot, ChannelId as AdapterChannelId}
 use domain::validator::message::{Heartbeat, Message, State, TYPE_HEARTBEAT};
 use domain::{Channel, ChannelId, RepositoryError, ValidatorId};
 
+use crate::application::MessagePropagator;
 use crate::domain::MessageRepository;
 
 pub struct HeartbeatFactory<A: Adapter> {
@@ -55,6 +56,7 @@ pub struct HeartbeatSender<A: Adapter> {
     message_repository: Box<dyn MessageRepository<A::State>>,
     adapter: A,
     factory: HeartbeatFactory<A>,
+    propagator: MessagePropagator<A::State>,
     // @TODO: Add config value for Heartbeat send frequency
 }
 
@@ -94,12 +96,13 @@ impl<A: Adapter> HeartbeatSender<A> {
         // call the HeartbeatFactory and create the new Heartbeat
         let heartbeat = await!(self.factory.create(signable_state_root.0))?;
 
-        // @TODO: Issue #93 - this should propagate the message to all validators!
-        // `add()` the heartbeat with the Repository
+        // Propagate the message to all Validators
+        // @TODO: Log errors
         await!(self
-            .message_repository
-            .add(&channel.id, &validator, Message::Heartbeat(heartbeat)))
-        .map_err(HeartbeatError::Repository)
+            .propagator
+            .propagate(&channel, Message::Heartbeat(heartbeat)));
+
+        Ok(())
     }
 
     fn is_heartbeat_time(&self, latest_heartbeat: &Heartbeat<A::State>) -> bool {

validator/src/application/message_propagation.rs

@@ -0,0 +1,151 @@
+use std::convert::TryFrom;
+use std::error::Error;
+use std::fmt;
+
+use domain::validator::message::{Message, State};
+use domain::{Channel, RepositoryError, ValidatorId};
+
+use crate::domain::MessageRepository;
+
+pub struct MessagePropagator<S: State> {
+    pub message_repository: Box<dyn MessageRepository<S>>,
+}
+
+#[derive(Debug)]
+pub enum PropagationErrorKind {
+    Repository(RepositoryError),
+}
+
+#[derive(Debug)]
+pub struct PropagationError {
+    kind: PropagationErrorKind,
+    message: String,
+}
+
+impl fmt::Display for PropagationError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.message)
+    }
+}
+
+impl Error for PropagationError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match &self.kind {
+            PropagationErrorKind::Repository(error) => Some(error),
+        }
+    }
+}
+
+impl From<RepositoryError> for PropagationError {
+    fn from(error: RepositoryError) -> Self {
+        Self {
+            kind: PropagationErrorKind::Repository(error),
+            message: "Repository call for propagating the message failed".to_string(),
+        }
+    }
+}
+
+impl<S: State> MessagePropagator<S> {
+    // @TODO: Make sure we have information for logging the results for particular Validator
+    pub async fn propagate<'a>(
+        &'a self,
+        channel: &'a Channel,
+        message: Message<S>,
+    ) -> Vec<Result<(), PropagationError>> {
+        let mut results = Vec::default();
+
+        for validator in channel.spec.validators.into_iter() {
+            // @TODO: Remove once we have ValidatorId in ValidatorDesc
+            let validator_id = ValidatorId::try_from(validator.id.as_str()).unwrap();
+            let add_result =
+                await!(self
+                    .message_repository
+                    .add(&channel.id, &validator_id, message.clone()))
+                .map_err(Into::into);
+            results.push(add_result);
+        }
+
+        results
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::cell::RefCell;
+
+    use futures::future::{ready, FutureExt};
+
+    use domain::channel::fixtures::get_channel;
+    use domain::validator::message::fixtures::{get_reject_state, DummyState};
+    use domain::validator::message::{Message, MessageType};
+    use domain::{ChannelId, ValidatorId};
+    use domain::{RepositoryError, RepositoryFuture};
+
+    use crate::application::MessagePropagator;
+    use crate::domain::MessageRepository;
+
+    struct MockMessageRepository<I>
+    where
+        I: Iterator<Item = Result<(), RepositoryError>>,
+    {
+        add_results: RefCell<I>,
+    }
+
+    impl<I> MessageRepository<DummyState> for MockMessageRepository<I>
+    where
+        I: Iterator<Item = Result<(), RepositoryError>>,
+    {
+        fn add(
+            &self,
+            _channel: &ChannelId,
+            _validator: &ValidatorId,
+            _message: Message<DummyState>,
+        ) -> RepositoryFuture<()> {
+            let result = self
+                .add_results
+                .borrow_mut()
+                .next()
+                .expect("Whoops, you called add() more than the provided results");
+            ready(result).boxed()
+        }
+
+        fn latest(
+            &self,
+            _channel: &ChannelId,
+            _from: &ValidatorId,
+            _types: Option<&[&MessageType]>,
+        ) -> RepositoryFuture<Option<Message<DummyState>>> {
+            unimplemented!("No need for latest in this Mock")
+        }
+    }
+
+    #[test]
+    fn propagates_and_returns_vector_of_results() {
+        futures::executor::block_on(async {
+            let add_error = RepositoryError::User;
+
+            let iterator = vec![Ok(()), Err(add_error)].into_iter();
+            let message_repository = MockMessageRepository {
+                add_results: RefCell::new(iterator),
+            };
+            let propagator = MessagePropagator {
+                message_repository: Box::new(message_repository),
+            };
+
+            let message = get_reject_state(None);
+            let channel = get_channel("id", &None, None);
+
+            let result = await!(propagator.propagate(&channel, Message::RejectState(message)));
+
+            assert_eq!(2, result.len());
+            assert!(result[0].is_ok());
+            match &result[1] {
+                Ok(_) => panic!("It should be an error"),
+                Err(error) => assert_eq!(
+                    "Repository call for propagating the message failed",
+                    error.message
+                ),
+            }
+        })
+    }
+}