Move request logic into its own module (#551)

Author: timorleph

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "aleph-bft"
-version = "0.44.0"
+version = "0.44.1"
 edition = "2021"
 authors = ["Cardinal Cryptography"]
 categories = ["algorithms", "data-structures", "cryptography", "database"]

consensus/Cargo.toml

@@ -18,7 +18,7 @@ pub type RecipientCountSchedule = Arc<dyn Fn(usize) -> usize + Sync + Send + 'st
 /// Configuration of several parameters related to delaying various tasks.
 #[derive(Clone)]
 pub struct DelayConfig {
-    /// Tick frequency of the Member. Governs internal task queue of the Member.
+    /// Tick frequency of the request task queue, impacts how often tasks are send out.
     pub tick_interval: Duration,
     /// Minimum frequency of broadcast of top known units. Units have to be at least this old to be
     /// rebroadcast at all.

consensus/src/config.rs

@@ -203,6 +203,11 @@ impl<H: Hasher, D: Data, MK: MultiKeychain> Dag<H, D, MK> {
         result
     }
 
+    /// The store of units currently being processed by this dag.
+    pub fn processing_units(&self) -> &UnitStore<SignedUnit<H, D, MK>> {
+        self.validator.processing_units()
+    }
+
     /// Notify the dag that a unit has finished processing and can be cleared from the cache.
     pub fn finished_processing(&mut self, hash: &H::Hash) {
         self.validator.finished_processing(hash);

consensus/src/dag/mod.rs

@@ -162,6 +162,11 @@ impl<H: Hasher, D: Data, MK: MultiKeychain> Validator<H, D, MK> {
         Ok(unit)
     }
 
+    /// The store of units currently being processed.
+    pub fn processing_units(&self) -> &UnitStore<SignedUnit<H, D, MK>> {
+        &self.processing_units
+    }
+
     /// Signal that a unit finished processing and thus it's copy no longer has to be kept for fork detection.
     /// NOTE: This is only a memory optimization, if the units stay there forever everything still works.
     pub fn finished_processing(&mut self, unit: &H::Hash) {

consensus/src/dag/validation.rs

@@ -10,7 +10,7 @@ mod responder;
 mod task;
 
 pub use responder::Responder;
-pub use task::DisseminationTask;
+pub use task::{Manager as TaskManager, ManagerStatus as TaskManagerStatus};
 
 /// Some form of message with the intended recipients.
 #[derive(Eq, PartialEq, Debug, Clone)]
@@ -33,6 +33,11 @@ impl<T> Addressed<T> {
         Addressed::new(message, vec![Recipient::Node(node_id)])
     }
 
+    /// Message that should be broadcast.
+    pub fn broadcast(message: T) -> Self {
+        Addressed::new(message, vec![Recipient::Everyone])
+    }
+
     /// All the recipients of this message.
     pub fn recipients(&self) -> &Vec<Recipient> {
         &self.recipients