http message service

Author: ghubertpalo

mithril-aggregator/src/database/provider/message.rs

@@ -11,11 +11,13 @@ use sqlite::ConnectionWithFullMutex;
 
 use super::{CertificateRecord, CertificateRecordProvider};
 
+/// Repository that turn inner Record entities into Message entities.
 pub struct CertificateMessageRepository {
     connection: Arc<ConnectionWithFullMutex>,
 }
 
 impl CertificateMessageRepository {
+    /// Constructor
     pub fn new(connection: Arc<ConnectionWithFullMutex>) -> Self {
         Self { connection }
     }
@@ -38,7 +40,7 @@ impl From<CertificateRecord> for CertificateMessage {
 
         CertificateMessage {
             hash: value.certificate_id,
-            previous_hash: value.parent_certificate_id.unwrap_or_else(|| String::new()),
+            previous_hash: value.parent_certificate_id.unwrap_or_default(),
             beacon: value.beacon,
             metadata,
             protocol_message: value.protocol_message,
@@ -59,15 +61,10 @@ impl From<CertificateRecord> for CertificateListItemMessage {
             sealed_at: value.sealed_at,
             total_signers: value.signers.len(),
         };
-        let (multi_signature, genesis_signature) = if value.parent_certificate_id.is_none() {
-            (String::new(), value.signature)
-        } else {
-            (value.signature, String::new())
-        };
 
         CertificateListItemMessage {
             hash: value.certificate_id,
-            previous_hash: value.parent_certificate_id.unwrap_or_else(|| String::new()),
+            previous_hash: value.parent_certificate_id.unwrap_or_default(),
             beacon: value.beacon,
             metadata,
             protocol_message: value.protocol_message,
@@ -78,13 +75,15 @@ impl From<CertificateRecord> for CertificateListItemMessage {
 }
 
 impl CertificateMessageRepository {
+    /// Return the certificate matching the given hash.
     pub async fn get_certificate(&self, hash: &str) -> StdResult<Option<CertificateMessage>> {
         let provider = CertificateRecordProvider::new(&self.connection);
         let mut cursor = provider.get_by_certificate_id(hash)?;
 
         Ok(cursor.next().map(|v| v.into()))
     }
 
+    /// Return the last N certificates
     pub async fn get_last(&self, limit: usize) -> StdResult<Vec<CertificateListItemMessage>> {
         let provider = CertificateRecordProvider::new(&self.connection);
         let cursor = provider.get_all()?;

mithril-aggregator/src/dependency_injection/builder.rs

@@ -1027,6 +1027,7 @@ impl DependenciesBuilder {
             ticker_service: self.get_ticker_service().await?,
             signed_entity_storer: self.get_signed_entity_storer().await?,
             signer_getter: self.get_signer_store().await?,
+            http_message_service: self.get_http_message_service().await?,
         };
 
         Ok(dependency_manager)

mithril-aggregator/src/dependency_injection/containers.rs

@@ -16,7 +16,7 @@ use mithril_common::{
     BeaconProvider,
 };
 
-use crate::services::EpochService;
+use crate::services::{EpochService, HttpMessageService};
 use crate::{
     configuration::*,
     database::provider::{CertificateRepository, SignedEntityStorer, SignerGetter, StakePoolStore},
@@ -132,6 +132,9 @@ pub struct DependencyContainer {
 
     /// Signer getter service
     pub signer_getter: Arc<dyn SignerGetter>,
+
+    /// HTTP message service
+    pub http_message_service: Arc<dyn HttpMessageService>,
 }
 
 #[doc(hidden)]

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

@@ -29,7 +29,7 @@ fn certificate_certificates(
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
     warp::path!("certificates")
         .and(warp::get())
-        .and(middlewares::with_certifier_service(dependency_manager))
+        .and(middlewares::with_http_message_service(dependency_manager))
         .and_then(handlers::certificate_certificates)
 }
 
@@ -39,16 +39,14 @@ fn certificate_certificate_hash(
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
     warp::path!("certificate" / String)
         .and(warp::get())
-        .and(middlewares::with_certifier_service(dependency_manager))
+        .and(middlewares::with_http_message_service(dependency_manager))
         .and_then(handlers::certificate_certificate_hash)
 }
 
 mod handlers {
     use crate::{
-        http_server::routes::reply,
-        message_adapters::{ToCertificateListMessageAdapter, ToCertificateMessageAdapter},
-        services::CertifierService,
-        CertificatePendingStore, ToCertificatePendingMessageAdapter,
+        http_server::routes::reply, services::HttpMessageService, CertificatePendingStore,
+        ToCertificatePendingMessageAdapter,
     };
 
     use mithril_common::messages::ToMessageAdapter;
@@ -80,18 +78,15 @@ mod handlers {
 
     /// List all Certificates
     pub async fn certificate_certificates(
-        certifier_service: Arc<dyn CertifierService>,
+        http_message_service: Arc<dyn HttpMessageService>,
     ) -> Result<impl warp::Reply, Infallible> {
         debug!("⇄ HTTP SERVER: certificate_certificates",);
 
-        match certifier_service
-            .get_latest_certificates(LIST_MAX_ITEMS)
+        match http_message_service
+            .get_last_certificates(LIST_MAX_ITEMS)
             .await
         {
-            Ok(certificates) => Ok(reply::json(
-                &ToCertificateListMessageAdapter::adapt(certificates),
-                StatusCode::OK,
-            )),
+            Ok(certificates) => Ok(reply::json(&certificates, StatusCode::OK)),
             Err(err) => {
                 warn!("certificate_certificates::error"; "error" => ?err);
                 Ok(reply::internal_server_error(err))
@@ -102,21 +97,18 @@ mod handlers {
     /// Certificate by certificate hash
     pub async fn certificate_certificate_hash(
         certificate_hash: String,
-        certifier_service: Arc<dyn CertifierService>,
+        http_message_service: Arc<dyn HttpMessageService>,
     ) -> Result<impl warp::Reply, Infallible> {
         debug!(
             "⇄ HTTP SERVER: certificate_certificate_hash/{}",
             certificate_hash
         );
 
-        match certifier_service
-            .get_certificate_by_hash(&certificate_hash)
+        match http_message_service
+            .get_certificate(&certificate_hash)
             .await
         {
-            Ok(Some(certificate)) => Ok(reply::json(
-                &ToCertificateMessageAdapter::adapt(certificate),
-                StatusCode::OK,
-            )),
+            Ok(Some(certificate)) => Ok(reply::json(&certificate, StatusCode::OK)),
             Ok(None) => Ok(reply::empty(StatusCode::NOT_FOUND)),
             Err(err) => {
                 warn!("certificate_certificate_hash::error"; "error" => ?err);

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

@@ -2,7 +2,7 @@ use crate::{
     database::provider::SignerGetter,
     dependency_injection::EpochServiceWrapper,
     event_store::{EventMessage, TransmitterService},
-    services::{CertifierService, SignedEntityService, TickerService},
+    services::{CertifierService, HttpMessageService, SignedEntityService, TickerService},
     CertificatePendingStore, Configuration, DependencyContainer, SignerRegisterer,
     VerificationKeyStorer,
 };
@@ -95,3 +95,10 @@ pub fn with_api_version_provider(
 ) -> impl Filter<Extract = (Arc<APIVersionProvider>,), Error = Infallible> + Clone {
     warp::any().map(move || dependency_manager.api_version_provider.clone())
 }
+
+/// With Message service
+pub fn with_http_message_service(
+    dependency_manager: Arc<DependencyContainer>,
+) -> impl Filter<Extract = (Arc<dyn HttpMessageService>,), Error = Infallible> + Clone {
+    warp::any().map(move || dependency_manager.http_message_service.clone())
+}

mithril-aggregator/src/services/http_message.rs

@@ -6,7 +6,7 @@ use async_trait::async_trait;
 use thiserror::Error;
 
 use mithril_common::{
-    messages::{CertificateListMessage, CertificateMessage, CertificatePendingMessage},
+    messages::{CertificateListMessage, CertificateMessage},
     StdResult,
 };
 
@@ -21,7 +21,7 @@ pub enum HttpMessageServiceError {
 }
 /// HTTP Message service trait.
 #[async_trait]
-pub trait HttpMessageService {
+pub trait HttpMessageService: Sync + Send {
     /// Return the message representation of a certificate if it exists.
     async fn get_certificate(
         &self,
@@ -30,16 +30,15 @@ pub trait HttpMessageService {
 
     /// Return the message representation of the last N certificates
     async fn get_last_certificates(&self, limit: usize) -> StdResult<CertificateListMessage>;
-
-    /// Return the message representation of the current pending certificate
-    async fn get_pending_certificate(&self) -> StdResult<CertificatePendingMessage>;
 }
 
+/// Implementation of the [HttpMessageService]
 pub struct MithrilHttpMessageService {
     certificate_message_repository: Arc<CertificateMessageRepository>,
 }
 
 impl MithrilHttpMessageService {
+    /// Constructor
     pub fn new(certificate_message_repository: Arc<CertificateMessageRepository>) -> Self {
         Self {
             certificate_message_repository,
@@ -61,10 +60,6 @@ impl HttpMessageService for MithrilHttpMessageService {
     async fn get_last_certificates(&self, limit: usize) -> StdResult<CertificateListMessage> {
         self.certificate_message_repository.get_last(limit).await
     }
-
-    async fn get_pending_certificate(&self) -> StdResult<CertificatePendingMessage> {
-        todo!()
-    }
 }
 
 #[cfg(test)]
@@ -73,8 +68,6 @@ mod tests {
 
     use crate::{dependency_injection::DependenciesBuilder, Configuration};
 
-    use super::*;
-
     #[tokio::test]
     async fn get_no_certificate() {
         // setup
@@ -130,6 +123,14 @@ mod tests {
             ..beacon.clone()
         };
         let genesis_certificate = fixture.create_genesis_certificate(&genesis_beacon);
+        dep_builder
+            .get_certificate_repository()
+            .await
+            .unwrap()
+            .create_certificate(genesis_certificate.clone())
+            .await
+            .unwrap();
+        let genesis_certificate = fixture.create_genesis_certificate(&beacon);
         dep_builder
             .get_certificate_repository()
             .await
@@ -140,6 +141,8 @@ mod tests {
 
         // test
         let certficate_messages = service.get_last_certificates(5).await.unwrap();
+
+        assert_eq!(2, certficate_messages.len());
         assert_eq!(genesis_certificate.hash, certficate_messages[0].hash);
     }
 }