Simplify json creation in query

Author: sfauvel

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

@@ -2,6 +2,7 @@ use crate::event_store::{Event, EventMessage};
 use chrono::Utc;
 use mithril_common::StdResult;
 use mithril_persistence::sqlite::{Query, SourceAlias, SqLiteEntity, WhereCondition};
+use serde_json::json;
 
 /// Queries to insert [EventMessage] to the sqlite database.
 pub struct InsertEventQuery {
@@ -15,11 +16,10 @@ impl InsertEventQuery {
             vec![
                 sqlite::Value::String(message.source),
                 sqlite::Value::String(message.action),
-                sqlite::Value::String(format!(
-                    r#"{{"headers": {}, "content": {}}}"#,
-                    serde_json::to_string(&message.headers)?,
-                    message.content
-                )),
+                sqlite::Value::String(serde_json::to_string(&json!({
+                    "content": message.content,
+                    "headers": message.headers,
+                }))?),
                 sqlite::Value::String(Utc::now().to_rfc3339()),
             ],
         );
@@ -48,13 +48,38 @@ mod tests {
 
     #[test]
     fn provider_sql() {
-        let message = EventMessage::new("source", "action", "content");
-        let (parameters, values) = InsertEventQuery::one(message).unwrap().filters().expand();
+        let message = EventMessage::new("source", "action", serde_json::json!("content"));
+
+        let (final_expression, parameters) =
+            InsertEventQuery::one(message).unwrap().filters().expand();
 
         assert_eq!(
             "(source, action, content, created_at) values (?1, ?2, ?3, ?4)".to_string(),
-            parameters
+            final_expression
+        );
+        assert_eq!(4, parameters.len());
+    }
+
+    #[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());
+
+        let (_, parameters) = InsertEventQuery::one(message).unwrap().filters().expand();
+
+        assert_eq!(
+            sqlite::Value::String(
+                r#"{"content":{"attr1":"content","attr2":123},"headers":{"key":"value"}}"#
+                    .to_string()
+            ),
+            parameters[2]
         );
-        assert_eq!(4, values.len());
     }
 }

mithril-aggregator/src/event_store/database/repository.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", "content");
+        let message = EventMessage::new("source", "action", serde_json::json!("content"));
 
         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", "content");
+        let message = EventMessage::new("source", "action", serde_json::json!("content"));
 
         let _event = persister.persist(message)?;
         Ok(())

mithril-aggregator/src/event_store/event.rs

@@ -16,29 +16,22 @@ pub struct EventMessage {
     pub action: String,
 
     /// JSON content of the message, its type is declared in the action property.
-    pub content: String,
+    pub content: serde_json::Value,
 
     /// Headers
     pub headers: HashMap<String, String>,
 }
 
 impl EventMessage {
     /// Instantiate a new EventMessage.
-    pub fn new(source: &str, action: &str, content: &str) -> Self {
+    pub fn new(source: &str, action: &str, content: serde_json::Value) -> Self {
         Self {
             source: source.to_string(),
             action: action.to_string(),
-            content: content.to_string(),
+            content,
             headers: HashMap::new(),
         }
     }
-
-    /// forge a new instance adding the given header
-    pub fn add_header(mut self, name: &str, value: &str) -> Self {
-        let _ = self.headers.insert(name.to_owned(), value.to_owned());
-
-        self
-    }
 }
 
 /// Event persisted in the Event Store.

mithril-aggregator/src/event_store/transmitter_service.rs

@@ -50,12 +50,8 @@ impl TransmitterService<EventMessage> {
     where
         T: Serialize,
     {
-        let content = serde_json::to_string(content).map_err(|e| {
-            let error_msg = format!("Serialization error while forging event message: {e}");
-            warn!(self.logger, "Event message error"; "error" => &error_msg);
+        let content = serde_json::json!(content);
 
-            error_msg
-        })?;
         let message = EventMessage {
             source: source.to_string(),
             action: action.to_string(),