Merge pull request #2089 from input-output-hk/sfa/2086/refactor_event_message_creation

Refactor event message creation

Author: sfauvel

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.5.103"
+version = "0.5.104"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/Cargo.toml

@@ -48,7 +48,7 @@ mod tests {
 
     #[test]
     fn provider_sql() {
-        let message = EventMessage::new("source", "action", serde_json::json!("content"));
+        let message = EventMessage::new("source", "action", &"content".to_string(), Vec::new());
 
         let (final_expression, parameters) =
             InsertEventQuery::one(message).unwrap().filters().expand();
@@ -62,15 +62,16 @@ mod tests {
 
     #[test]
     fn build_a_json_for_content_field_with_content_and_headers() {
-        let content = serde_json::json!({
-            "attr1": "content".to_string(),
-            "attr2": 123,
-        });
-
-        let mut message = EventMessage::new("source", "action", content);
-        message
-            .headers
-            .insert("key".to_string(), "value".to_string());
+        #[derive(serde::Serialize)]
+        struct Content {
+            attr1: String,
+            attr2: i32,
+        }
+        let content = Content {
+            attr1: "content".to_string(),
+            attr2: 123,
+        };
+        let message = EventMessage::new("source", "action", &content, [("key", "value")].to_vec());
 
         let (_, parameters) = InsertEventQuery::one(message).unwrap().filters().expand();
 

mithril-aggregator/src/event_store/database/query.rs

@@ -42,7 +42,7 @@ mod tests {
         let connection = Arc::new(event_store_db_connection().unwrap());
 
         let persister = EventPersister::new(connection);
-        let message = EventMessage::new("source", "action", serde_json::json!("content"));
+        let message = EventMessage::new("source", "action", &"content".to_string(), Vec::new());
 
         let _event = persister.persist(message)?;
         Ok(())
@@ -53,7 +53,7 @@ mod tests {
         let connection = Arc::new(event_store_db_connection().unwrap());
 
         let persister = EventPersister::new(connection);
-        let message = EventMessage::new("source", "action", serde_json::json!("content"));
+        let message = EventMessage::new("source", "action", &"content".to_string(), Vec::new());
 
         let _event = persister.persist(message)?;
         Ok(())
@@ -63,9 +63,7 @@ mod tests {
         use std::time::Duration;
 
         use crate::{
-            event_store::{database::test_helper::event_store_db_connection, TransmitterService},
-            services::UsageReporter,
-            test_tools::TestLogger,
+            event_store::database::test_helper::event_store_db_connection, services::UsageReporter,
         };
         use chrono::DateTime;
 
@@ -103,18 +101,13 @@ mod tests {
             let metric_date =
                 DateTime::parse_from_str(&format!("{date} +0000"), "%Y-%m-%d %H:%M:%S %z").unwrap();
 
-            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<EventMessage>();
-            let transmitter_service = Arc::new(TransmitterService::new(tx, TestLogger::stdout()));
-            let _result = UsageReporter::send_metric_event(
-                &transmitter_service,
+            let message = UsageReporter::create_metrics_event_message(
                 metric_name.to_string(),
                 value,
                 Duration::from_secs(5),
                 metric_date.into(),
             );
 
-            let message: EventMessage = rx.try_recv().unwrap();
-
             let _event = persister.persist(message).unwrap();
         }
 
@@ -150,12 +143,9 @@ mod tests {
     mod signer_registration_summary {
         use std::sync::Arc;
 
-        use crate::event_store::{
-            database::test_helper::event_store_db_connection, TransmitterService,
-        };
-        use crate::test_tools::TestLogger;
-        use mithril_common::entities::Stake;
-        use mithril_common::{entities::SignerWithStake, test_utils::fake_data, StdResult};
+        use crate::event_store::database::test_helper::event_store_db_connection;
+        use mithril_common::entities::{SignerWithStake, Stake};
+        use mithril_common::{test_utils::fake_data, StdResult};
         use sqlite::ConnectionThreadSafe;
 
         use super::{EventMessage, EventPersister};
@@ -168,25 +158,20 @@ mod tests {
             stake: Stake,
             signer_node_version: &str,
         ) {
-            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<EventMessage>();
-            let transmitter_service = Arc::new(TransmitterService::new(tx, TestLogger::stdout()));
-
             let signers = fake_data::signers_with_stakes(1);
             let signer = SignerWithStake {
                 party_id: party_id.to_string(),
                 stake,
                 ..signers[0].clone()
             };
 
-            let _ = transmitter_service.send_signer_registration_event(
+            let message = EventMessage::signer_registration(
                 "Test",
                 &signer,
                 Some(signer_node_version.to_string()),
                 epoch,
             );
 
-            let message: EventMessage = rx.try_recv().unwrap();
-
             let _event = persister.persist(message).unwrap();
         }
 

mithril-aggregator/src/event_store/database/repository.rs

@@ -1,4 +1,6 @@
 use chrono::{DateTime, Utc};
+use mithril_common::entities::SignerWithStake;
+use serde::Serialize;
 
 use std::collections::HashMap;
 
@@ -24,13 +26,40 @@ pub struct EventMessage {
 
 impl EventMessage {
     /// Instantiate a new EventMessage.
-    pub fn new(source: &str, action: &str, content: serde_json::Value) -> Self {
-        Self {
+    pub fn new<T>(source: &str, action: &str, content: &T, headers: Vec<(&str, &str)>) -> Self
+    where
+        T: Serialize,
+    {
+        let content = serde_json::json!(content);
+
+        EventMessage {
             source: source.to_string(),
             action: action.to_string(),
             content,
-            headers: HashMap::new(),
+            headers: headers
+                .into_iter()
+                .map(|(h, v)| (h.to_string(), v.to_string()))
+                .collect(),
+        }
+    }
+
+    /// Create a signer registration event message.
+    pub fn signer_registration(
+        source: &str,
+        signer_with_stake: &SignerWithStake,
+        signer_node_version: Option<String>,
+        epoch_str: &str,
+    ) -> Self {
+        let mut headers: Vec<(&str, &str)> = match signer_node_version.as_ref() {
+            Some(version) => vec![("signer-node-version", version)],
+            None => Vec::new(),
+        };
+
+        if !epoch_str.is_empty() {
+            headers.push(("epoch", epoch_str));
         }
+
+        Self::new::<SignerWithStake>(source, "register_signer", signer_with_stake, headers)
     }
 }
 

mithril-aggregator/src/event_store/event.rs

@@ -1,9 +1,9 @@
-use serde::Serialize;
+use anyhow::Context;
 use slog::{warn, Logger};
 use std::fmt::Debug;
 use tokio::sync::mpsc::UnboundedSender;
 
-use mithril_common::{entities::SignerWithStake, logging::LoggerExtensions};
+use mithril_common::{logging::LoggerExtensions, StdResult};
 
 use super::EventMessage;
 
@@ -36,62 +36,22 @@ where
 }
 
 impl TransmitterService<EventMessage> {
-    /// Craft and send an [EventMessage] given the serializable data.
-    /// This method is done in a way to make as simple as possible to send a
-    /// message and make any error not to cause a business failure. A warning is
-    /// issued so the resulting error may be discarded.
-    pub fn send_event_message<T>(
-        &self,
-        source: &str,
-        action: &str,
-        content: &T,
-        headers: Vec<(&str, &str)>,
-    ) -> Result<(), String>
-    where
-        T: Serialize,
-    {
-        let content = serde_json::json!(content);
-
-        let message = EventMessage {
-            source: source.to_string(),
-            action: action.to_string(),
-            content,
-            headers: headers
-                .into_iter()
-                .map(|(h, v)| (h.to_string(), v.to_string()))
-                .collect(),
-        };
-        self.get_transmitter().send(message.clone()).map_err(|e| {
-            let error_msg =
-                format!("An error occurred when sending message {message:?} to monitoring: '{e}'.");
-            warn!(self.logger, "Event message error"; "error" => &error_msg);
-
-            error_msg
-        })
+    /// Send an [EventMessage].
+    pub fn try_send(&self, message: EventMessage) -> StdResult<()> {
+        self.get_transmitter()
+            .send(message.clone())
+            .with_context(|| {
+                format!("An error occurred when sending message {message:?} to monitoring.")
+            })
     }
 
-    /// Send signer registration event.
-    pub fn send_signer_registration_event(
-        &self,
-        source: &str,
-        signer_with_stake: &SignerWithStake,
-        signer_node_version: Option<String>,
-        epoch_str: &str,
-    ) -> Result<(), String> {
-        let mut headers: Vec<(&str, &str)> = match signer_node_version.as_ref() {
-            Some(version) => vec![("signer-node-version", version)],
-            None => Vec::new(),
+    /// Send an [EventMessage].
+    ////
+    /// An error when sending a message has no impact on the business.
+    /// If there is one, a warning is issued so the resulting error may be safely ignored by the caller.
+    pub fn send(&self, message: EventMessage) {
+        if let Err(e) = self.try_send(message.clone()) {
+            warn!(self.logger, "Event message error"; "error" => ?e);
         };
-
-        if !epoch_str.is_empty() {
-            headers.push(("epoch", epoch_str));
-        }
-
-        self.send_event_message::<SignerWithStake>(
-            source,
-            "register_signer",
-            signer_with_stake,
-            headers,
-        )
     }
 }

mithril-aggregator/src/event_store/transmitter_service.rs

@@ -144,24 +144,25 @@ mod handlers {
             .await
         {
             Ok(signer_with_stake) => {
-                let _ = event_transmitter.send_signer_registration_event(
+                event_transmitter.send(EventMessage::signer_registration(
                     "HTTP::signer_register",
                     &signer_with_stake,
                     signer_node_version,
                     epoch_str.as_str(),
-                );
+                ));
 
                 Ok(reply::empty(StatusCode::CREATED))
             }
             Err(SignerRegistrationError::ExistingSigner(signer_with_stake)) => {
                 debug!(logger, "register_signer::already_registered");
 
-                let _ = event_transmitter.send_signer_registration_event(
+                event_transmitter.send(EventMessage::signer_registration(
                     "HTTP::signer_register",
                     &signer_with_stake,
                     signer_node_version,
                     epoch_str.as_str(),
-                );
+                ));
+
                 Ok(reply::empty(StatusCode::CREATED))
             }
             Err(SignerRegistrationError::FailedSignerRegistration(err)) => {

mithril-aggregator/src/http_server/routes/signer_routes.rs

@@ -16,12 +16,14 @@ fn post_statistics(
     warp::path!("statistics" / "snapshot")
         .and(warp::post())
         .and(warp::body::json())
+        .and(middlewares::with_logger(router_state))
         .and(middlewares::with_event_transmitter(router_state))
         .and(middlewares::with_metrics_service(router_state))
         .and_then(handlers::post_snapshot_statistics)
 }
 
 mod handlers {
+    use slog::warn;
     use std::{convert::Infallible, sync::Arc};
     use warp::http::StatusCode;
 
@@ -33,6 +35,7 @@ mod handlers {
 
     pub async fn post_snapshot_statistics(
         snapshot_download_message: SnapshotDownloadMessage,
+        logger: slog::Logger,
         event_transmitter: Arc<TransmitterService<EventMessage>>,
         metrics_service: Arc<MetricsService>,
     ) -> Result<impl warp::Reply, Infallible> {
@@ -42,13 +45,18 @@ mod handlers {
 
         let headers: Vec<(&str, &str)> = Vec::new();
 
-        match event_transmitter.send_event_message(
+        let message = EventMessage::new(
             "HTTP::statistics",
             "snapshot_downloaded",
             &snapshot_download_message,
             headers,
-        ) {
-            Err(e) => Ok(reply::internal_server_error(e)),
+        );
+
+        match event_transmitter.try_send(message.clone()) {
+            Err(e) => {
+                warn!(logger, "Event message error"; "error" => ?e);
+                Ok(reply::internal_server_error(e))
+            }
             Ok(_) => Ok(reply::empty(StatusCode::CREATED)),
         }
     }