refactor(aggregator): add a stop signal channel in DI

Author: jpraynaud

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

@@ -81,9 +81,11 @@ impl DependenciesBuilder {
 
     /// Builds a [SignatureProcessor]
     pub async fn create_signature_processor(&mut self) -> Result<Arc<dyn SignatureProcessor>> {
+        let (_stop_tx, stop_rx) = self.get_stop_signal_channel().await?;
         let signature_processor = SequentialSignatureProcessor::new(
             self.build_signature_consumer().await?,
             self.get_certifier_service().await?,
+            stop_rx,
             self.root_logger(),
         );
 

mithril-aggregator/src/dependency_injection/builder/mod.rs

@@ -8,7 +8,7 @@ use std::{path::PathBuf, sync::Arc};
 use tokio::{
     sync::{
         mpsc::{UnboundedReceiver, UnboundedSender},
-        Mutex,
+        watch, Mutex,
     },
     time::Duration,
 };
@@ -273,6 +273,9 @@ pub struct DependenciesBuilder {
 
     /// Protocol parameters retriever
     pub protocol_parameters_retriever: Option<Arc<dyn ProtocolParametersRetriever>>,
+
+    /// Stop signal channel
+    pub stop_signal_channel: Option<(watch::Sender<()>, watch::Receiver<()>)>,
 }
 
 impl DependenciesBuilder {
@@ -335,6 +338,7 @@ impl DependenciesBuilder {
             metrics_service: None,
             leader_aggregator_client: None,
             protocol_parameters_retriever: None,
+            stop_signal_channel: None,
         }
     }
 

mithril-aggregator/src/dependency_injection/builder/support/mod.rs

@@ -4,6 +4,7 @@
 
 mod compatibility;
 mod observability;
+mod signal;
 mod sqlite;
 mod stores;
 mod upkeep;

mithril-aggregator/src/dependency_injection/builder/support/signal.rs

@@ -0,0 +1,20 @@
+use tokio::sync::watch;
+
+use crate::dependency_injection::{DependenciesBuilder, Result};
+use crate::get_dependency;
+
+impl DependenciesBuilder {
+    /// Builds a stop signal channel
+    pub async fn build_stop_signal_channel(
+        &mut self,
+    ) -> Result<(watch::Sender<()>, watch::Receiver<()>)> {
+        Ok(watch::channel(()))
+    }
+
+    /// Get the stop signal channel
+    pub async fn get_stop_signal_channel(
+        &mut self,
+    ) -> Result<(watch::Sender<()>, watch::Receiver<()>)> {
+        get_dependency!(self.stop_signal_channel)
+    }
+}