feat(aggregator): wire 'SignatureConsumerDmq' in DI

Author: jpraynaud

mithril-aggregator/src/configuration.rs

@@ -61,12 +61,16 @@ pub trait ConfigurationSource {
         panic!("cardano_cli_path is not implemented.");
     }
 
-    /// Path of the socket used by the Cardano CLI tool
-    /// to communicate with the Cardano node
+    /// Path of the socket opened by the Cardano node
     fn cardano_node_socket_path(&self) -> PathBuf {
         panic!("cardano_node_socket_path is not implemented.");
     }
 
+    /// Path of the socket opened by the DMQ node
+    fn dmq_node_socket_path(&self) -> Option<PathBuf> {
+        panic!("dmq_node_socket_path is not implemented.");
+    }
+
     /// Cardano node version.
     ///
     /// **NOTE**: This cannot be verified for now (see [this
@@ -390,11 +394,14 @@ pub struct ServeCommandConfiguration {
     #[example = "`cardano-cli`"]
     pub cardano_cli_path: PathBuf,
 
-    /// Path of the socket used by the Cardano CLI tool
-    /// to communicate with the Cardano node
-    #[example = "`/tmp/cardano.sock`"]
+    /// Path of the socket opened by the Cardano node
+    #[example = "`/ipc/node.socket`"]
     pub cardano_node_socket_path: PathBuf,
 
+    /// Path of the socket opened by the DMQ node
+    #[example = "`/ipc/dmq.socket`"]
+    pub dmq_node_socket_path: Option<PathBuf>,
+
     /// Cardano node version.
     ///
     /// **NOTE**: This cannot be verified for now (see [this
@@ -628,6 +635,7 @@ impl ServeCommandConfiguration {
             environment: ExecutionEnvironment::Test,
             cardano_cli_path: PathBuf::new(),
             cardano_node_socket_path: PathBuf::new(),
+            dmq_node_socket_path: None,
             cardano_node_version: "0.0.1".to_string(),
             network_magic: Some(42),
             network: "devnet".to_string(),
@@ -707,6 +715,10 @@ impl ConfigurationSource for ServeCommandConfiguration {
         self.cardano_node_socket_path.clone()
     }
 
+    fn dmq_node_socket_path(&self) -> Option<PathBuf> {
+        self.dmq_node_socket_path.clone()
+    }
+
     fn cardano_node_version(&self) -> String {
         self.cardano_node_version.clone()
     }

mithril-aggregator/src/dependency_injection/builder/enablers/misc.rs

@@ -7,14 +7,17 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use mithril_common::messages::RegisterSignatureMessageDmq;
+use mithril_dmq_node::DmqConsumerPallas;
 use mithril_signed_entity_lock::SignedEntityTypeLock;
 
 use crate::database::repository::CertificateRepository;
 use crate::dependency_injection::{DependenciesBuilder, Result};
 use crate::get_dependency;
 use crate::services::{
     AggregatorClient, AggregatorHTTPClient, MessageService, MithrilMessageService,
-    SequentialSignatureProcessor, SignatureConsumer, SignatureConsumerNoop, SignatureProcessor,
+    SequentialSignatureProcessor, SignatureConsumer, SignatureConsumerDmq, SignatureConsumerNoop,
+    SignatureProcessor,
 };
 impl DependenciesBuilder {
     async fn build_signed_entity_type_lock(&mut self) -> Result<Arc<SignedEntityTypeLock>> {
@@ -74,9 +77,19 @@ impl DependenciesBuilder {
 
     /// Builds a [SignatureConsumer]
     pub async fn build_signature_consumer(&mut self) -> Result<Arc<dyn SignatureConsumer>> {
-        let signature_consumer = SignatureConsumerNoop;
+        let signature_consumer =
+            if let Some(dmq_node_socket_path) = self.configuration.dmq_node_socket_path() {
+                let dmq_consumer = Arc::new(DmqConsumerPallas::<RegisterSignatureMessageDmq>::new(
+                    dmq_node_socket_path,
+                    self.configuration.get_network()?,
+                    self.root_logger(),
+                ));
+                Arc::new(SignatureConsumerDmq::new(dmq_consumer)) as Arc<dyn SignatureConsumer>
+            } else {
+                Arc::new(SignatureConsumerNoop) as Arc<dyn SignatureConsumer>
+            };
 
-        Ok(Arc::new(signature_consumer))
+        Ok(signature_consumer)
     }
 
     /// Builds a [SignatureProcessor]