feat(dmq): update aggregator relay for DMQ messages

Author: jpraynaud

mithril-relay/src/relay/aggregator.rs

@@ -1,35 +1,83 @@
-use crate::p2p::{BroadcastMessage, Peer, PeerEvent};
+use std::{path::Path, sync::Arc};
+
 use anyhow::anyhow;
 use libp2p::Multiaddr;
+use mithril_dmq::{DmqConsumerServer, DmqConsumerServerPallas, DmqMessage};
+use reqwest::StatusCode;
+use slog::{Logger, error, info};
+use tokio::sync::{
+    mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
+    watch::{self, Receiver},
+};
+
 use mithril_common::{
-    StdResult,
+    CardanoNetwork, StdResult,
     logging::LoggerExtensions,
     messages::{RegisterSignatureMessageHttp, RegisterSignerMessage},
 };
-use reqwest::StatusCode;
-use slog::{Logger, error, info};
+
+use crate::p2p::{BroadcastMessage, Peer, PeerEvent};
 
 /// A relay for a Mithril aggregator
 pub struct AggregatorRelay {
     aggregator_endpoint: String,
     peer: Peer,
+    signature_dmq_tx: UnboundedSender<DmqMessage>,
     logger: Logger,
 }
 
 impl AggregatorRelay {
     /// Start a relay for a Mithril aggregator
     pub async fn start(
         addr: &Multiaddr,
+        dmq_node_socket_path: &Path,
+        cardano_network: &CardanoNetwork,
         aggregator_endpoint: &str,
         logger: &Logger,
     ) -> StdResult<Self> {
+        let (_stop_tx, stop_rx) = watch::channel(());
+        let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
+        let _dmq_consumer_server = Self::start_dmq_consumer_server(
+            dmq_node_socket_path,
+            cardano_network,
+            signature_dmq_rx,
+            stop_rx,
+            logger.clone(),
+        )
+        .await?;
+
         Ok(Self {
             aggregator_endpoint: aggregator_endpoint.to_owned(),
             peer: Peer::new(addr).with_logger(logger).start().await?,
+            signature_dmq_tx,
             logger: logger.new_with_component_name::<Self>(),
         })
     }
 
+    async fn start_dmq_consumer_server(
+        socket: &Path,
+        cardano_network: &CardanoNetwork,
+        signature_dmq_rx: UnboundedReceiver<DmqMessage>,
+        stop_rx: Receiver<()>,
+        logger: Logger,
+    ) -> StdResult<Arc<DmqConsumerServerPallas>> {
+        let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
+            socket.to_path_buf(),
+            cardano_network.to_owned(),
+            stop_rx,
+            logger.clone(),
+        ));
+        dmq_consumer_server.register_receiver(signature_dmq_rx).await?;
+        let dmq_consumer_server_clone = dmq_consumer_server.clone();
+        tokio::spawn(async move {
+            if let Err(err) = dmq_consumer_server_clone.run().await {
+                error!(logger.to_owned(), "DMQ Consumer server failed"; "error" => ?err);
+            }
+        });
+
+        Ok(dmq_consumer_server)
+    }
+
     async fn notify_signature_to_aggregator(
         &self,
         signature_message: &RegisterSignatureMessageHttp,
@@ -100,7 +148,7 @@ impl AggregatorRelay {
     pub async fn tick(&mut self) -> StdResult<()> {
         if let Some(peer_event) = self.peer.tick_swarm().await? {
             match self.peer.convert_peer_event_to_message(peer_event) {
-                Ok(Some(BroadcastMessage::RegisterSigner(signer_message_received))) => {
+                Ok(Some(BroadcastMessage::RegisterSignerHttp(signer_message_received))) => {
                     let retry_max = 3;
                     let mut retry_count = 0;
                     while let Err(e) =
@@ -113,7 +161,7 @@ impl AggregatorRelay {
                         }
                     }
                 }
-                Ok(Some(BroadcastMessage::RegisterSignature(signature_message_received))) => {
+                Ok(Some(BroadcastMessage::RegisterSignatureHttp(signature_message_received))) => {
                     let retry_max = 3;
                     let mut retry_count = 0;
                     while let Err(e) =
@@ -126,6 +174,11 @@ impl AggregatorRelay {
                         }
                     }
                 }
+                Ok(Some(BroadcastMessage::RegisterSignatureDmq(signature_message_received))) => {
+                    self.signature_dmq_tx.send(signature_message_received).map_err(|e| {
+                        anyhow!("Failed to send signature message to DMQ consumer server: {e}")
+                    })?;
+                }
                 Ok(None) => {}
                 Err(e) => return Err(e),
             }
@@ -179,9 +232,15 @@ mod tests {
             then.status(201).body("ok");
         });
         let addr: Multiaddr = "/ip4/0.0.0.0/tcp/0".parse().unwrap();
-        let relay = AggregatorRelay::start(&addr, &server.url(""), &TestLogger::stdout())
-            .await
-            .unwrap();
+        let relay = AggregatorRelay::start(
+            &addr,
+            &Path::new("test"),
+            &CardanoNetwork::TestNet(123),
+            &server.url(""),
+            &TestLogger::stdout(),
+        )
+        .await
+        .unwrap();
 
         relay
             .notify_signature_to_aggregator(&RegisterSignatureMessageHttp::dummy())