Issue #74 Validator MemoryMessageRepository (#92)

* validator - main - use the MemoryChannelRepository by default, but leave the API instantiation as well

* domain - validator - message - make all fields `pub` & add todo for fixture

* [test] validator - persistence - validator - memory - `add()` & 1 case of `latest()`

* [test] validator - persistence - validator - memory - `latest` filtering by `type` & `from`

Author: elpiel

domain/src/validator/message.rs

@@ -60,23 +60,23 @@ pub const ALL_TYPES: [&MessageType; 5] = [
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct ApproveState<S: State> {
-    state_root: S::StateRoot,
-    signature: S::Signature,
-    is_healthy: bool,
+    pub state_root: S::StateRoot,
+    pub signature: S::Signature,
+    pub is_healthy: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct NewState<S: State> {
-    state_root: S::StateRoot,
-    signature: S::Signature,
-    balances: BalancesMap,
+    pub state_root: S::StateRoot,
+    pub signature: S::Signature,
+    pub balances: BalancesMap,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct RejectState {
-    reason: String,
+    pub reason: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -105,10 +105,10 @@ impl<S: State> Heartbeat<S> {
 #[serde(rename_all = "camelCase")]
 pub struct Accounting {
     #[serde(rename = "last_ev_aggr")]
-    last_event_aggregate: DateTime<Utc>,
+    pub last_event_aggregate: DateTime<Utc>,
     #[serde(rename = "balances_pre_fees")]
-    pre_fees: BalancesMap,
-    balances: BalancesMap,
+    pub pre_fees: BalancesMap,
+    pub balances: BalancesMap,
 }
 
 #[cfg(any(test, feature = "fixtures"))]

validator/src/application/heartbeat.rs

@@ -62,7 +62,7 @@ impl<A: Adapter + State> HeartbeatSender<A> {
         let validator = ValidatorId::try_from(self.adapter.config().identity.as_ref()).unwrap();
         let latest_future =
             self.message_repository
-                .latest(channel.id, validator, Some(&[&TYPE_HEARTBEAT]));
+                .latest(&channel.id, &validator, Some(&[&TYPE_HEARTBEAT]));
         let latest_heartbeat = await!(latest_future)
             .map_err(HeartbeatError::Repository)?
             .map(|heartbeat_msg| match heartbeat_msg {

validator/src/domain/validator.rs

@@ -27,8 +27,8 @@ pub mod repository {
 
         fn latest(
             &self,
-            channel_id: ChannelId,
-            from: ValidatorId,
+            channel_id: &ChannelId,
+            from: &ValidatorId,
             types: Option<&[&MessageType]>,
         ) -> RepositoryFuture<Option<Message<S>>>;
     }

validator/src/infrastructure/persistence/validator/memory.rs

@@ -20,8 +20,8 @@ impl State for MemoryState {
 #[derive(Clone)]
 pub struct MemoryMessage {
     pub message: Message<MemoryState>,
-    pub channel_id: ChannelId,
-    pub validator_id: ValidatorId,
+    pub channel: ChannelId,
+    pub owner: ValidatorId,
 }
 
 pub struct MemoryMessageRepository {
@@ -46,8 +46,8 @@ impl MessageRepository<MemoryState> for MemoryMessageRepository {
     fn add(&self, channel_id: ChannelId, message: Message<MemoryState>) -> RepositoryFuture<()> {
         let message = MemoryMessage {
             message,
-            channel_id,
-            validator_id: self.self_validator_id.clone(),
+            channel: channel_id,
+            owner: self.self_validator_id.clone(),
         };
         // this should never match against the new record, that's why always pass false.
         ready(self.inner.add(&false, message).map_err(Into::into)).boxed()
@@ -58,15 +58,15 @@ impl MessageRepository<MemoryState> for MemoryMessageRepository {
     /// If not `types` are provided, it will match against all types.
     fn latest(
         &self,
-        channel_id: ChannelId,
-        from: ValidatorId,
+        channel_id: &ChannelId,
+        from: &ValidatorId,
         types: Option<&[&MessageType]>,
     ) -> RepositoryFuture<Option<Message<MemoryState>>> {
         let latest = self
             .inner
             .list_all(|mem_msg| {
-                let is_from = mem_msg.validator_id == from;
-                let is_channel_id = mem_msg.channel_id == channel_id;
+                let is_from = &mem_msg.owner == from;
+                let is_channel_id = &mem_msg.channel == channel_id;
                 // if there are no types provided, it should match every type, i.e. default `true` for `None`
                 let is_in_types = types.map_or(true, |message_types| {
                     mem_msg.message.is_types(message_types)
@@ -82,3 +82,148 @@ impl MessageRepository<MemoryState> for MemoryMessageRepository {
         ready(latest.map_err(Into::into)).boxed()
     }
 }
+
+#[cfg(test)]
+mod test {
+    use std::convert::TryFrom;
+
+    use domain::fixtures::get_channel_id;
+    use domain::validator::message::fixtures::{get_heartbeat, get_reject_state};
+
+    use super::*;
+    use domain::validator::message::TYPE_REJECT;
+
+    fn get_reject_memory_message(
+        channel: &ChannelId,
+        owner: &ValidatorId,
+        reject_reason: Option<String>,
+    ) -> MemoryMessage {
+        MemoryMessage {
+            channel: channel.clone(),
+            owner: owner.clone(),
+            message: Message::RejectState(get_reject_state(reject_reason)),
+        }
+    }
+
+    #[test]
+    fn adds_message_with_the_self_validator_id() {
+        futures::executor::block_on(async {
+            let validator_id = ValidatorId::try_from("identity").expect("ValidatorId failed");
+            let repo = MemoryMessageRepository::new(&[], validator_id.clone());
+
+            let message = get_reject_state(None);
+            let channel_id = get_channel_id("channel id");
+
+            await!(repo.add(channel_id, Message::RejectState(message)))
+                .expect("Adding a message failed");
+
+            let list_all = repo
+                .inner
+                .list_all(|m| Some(m.clone()))
+                .expect("Listing all Messages failed");
+
+            assert_eq!(1, list_all.len());
+            assert_eq!(validator_id, list_all[0].owner);
+            assert_eq!(channel_id, list_all[0].channel);
+        })
+    }
+
+    #[test]
+    fn getting_latest_message_with_self_validator_id() {
+        futures::executor::block_on(async {
+            let validator_id = ValidatorId::try_from("identity").expect("ValidatorId failed");
+            let channel_id = get_channel_id("channel id");
+
+            let repo = MemoryMessageRepository::new(&[], validator_id.clone());
+            // add an initial Reject message for checking latest ordering
+            let init_message =
+                Message::RejectState(get_reject_state(Some("Initial Message".to_string())));
+            await!(repo.add(channel_id.clone(), init_message))
+                .expect("Adding the initial message failed");
+
+            let new_message = Message::RejectState(get_reject_state(Some("my reason".to_string())));
+            await!(repo.add(channel_id.clone(), new_message)).expect("Adding a message failed");
+
+            let latest_any = await!(repo.latest(&channel_id, &validator_id, None))
+                .expect("Getting latest Message failed");
+
+            match latest_any.expect("There was no latest message returned") {
+                Message::RejectState(reject_state) => assert_eq!("my reason", reject_state.reason),
+                _ => panic!("A Reject state message was not returned as latest message!"),
+            }
+        })
+    }
+
+    #[test]
+    fn getting_latest_message_filters_by_from() {
+        futures::executor::block_on(async {
+            let self_validator_id = ValidatorId::try_from("identity").expect("ValidatorId failed");
+            let channel = get_channel_id("channel id");
+
+            let from =
+                ValidatorId::try_from("another validator").expect("ValidatorId for form failed");
+            let init_message = get_reject_memory_message(&channel, &self_validator_id, None);
+
+            let repo = MemoryMessageRepository::new(&[init_message], self_validator_id.clone());
+
+            let result =
+                await!(repo.latest(&channel, &from, None)).expect("Fetching latest message failed");
+
+            assert!(
+                result.is_none(),
+                "A latest message was found, even though it has different ValidatorId"
+            );
+        })
+    }
+
+    #[test]
+    fn getting_latest_message_filters_by_channel_id() {
+        futures::executor::block_on(async {
+            let self_validator_id = ValidatorId::try_from("identity").expect("ValidatorId failed");
+            let channel = get_channel_id("channel 1");
+            let from_channel = get_channel_id("channel 2");
+
+            let init_message = get_reject_memory_message(&channel, &self_validator_id, None);
+
+            let repo = MemoryMessageRepository::new(&[init_message], self_validator_id.clone());
+
+            let result = await!(repo.latest(&from_channel, &self_validator_id, None))
+                .expect("Fetching latest message failed");
+
+            assert!(
+                result.is_none(),
+                "A latest message was found, even though it has different ChannelId"
+            );
+        })
+    }
+
+    #[test]
+    fn getting_latest_message_filters_by_types() {
+        futures::executor::block_on(async {
+            let self_validator_id = ValidatorId::try_from("identity").expect("ValidatorId failed");
+            let channel = get_channel_id("channel 1");
+
+            let init_messages = [
+                get_reject_memory_message(&channel, &self_validator_id, Some("reason".to_string())),
+                MemoryMessage {
+                    message: Message::Heartbeat(get_heartbeat::<MemoryState>(
+                        "state".to_string(),
+                        "signature".to_string(),
+                    )),
+                    channel: channel.clone(),
+                    owner: self_validator_id.clone(),
+                },
+            ];
+
+            let repo = MemoryMessageRepository::new(&init_messages, self_validator_id.clone());
+
+            let result = await!(repo.latest(&channel, &self_validator_id, Some(&[&TYPE_REJECT])))
+                .expect("Fetching latest message failed");
+
+            match result.expect("There was no latest message returned") {
+                Message::RejectState(reject_state) => assert_eq!("reason", reject_state.reason),
+                _ => panic!("Filtering by type didn't return the expected RejectState message!"),
+            }
+        })
+    }
+}

validator/src/main.rs

@@ -82,15 +82,18 @@ fn run(is_single_tick: bool, adapter: impl Adapter) {
     use validator::application::validator::{Follower, Leader};
     use validator::application::worker::{InfiniteWorker, TickWorker};
     use validator::domain::worker::Worker;
-    use validator::infrastructure::persistence::channel::api::ApiChannelRepository;
+    use validator::infrastructure::persistence::channel::{
+        ApiChannelRepository, MemoryChannelRepository,
+    };
     use validator::infrastructure::sentry::SentryApi;
 
     let sentry = SentryApi {
         client: Client::new(),
         sentry_url: CONFIG.sentry_url.clone(),
     };
 
-    let channel_repository = Arc::new(ApiChannelRepository { sentry });
+    let _channel_repository = Arc::new(ApiChannelRepository { sentry });
+    let channel_repository = Arc::new(MemoryChannelRepository::new(&[]));
 
     let tick_worker = TickWorker {
         leader: Leader {},