feat(consensus): define marshal-specific reporter trait

Author: SuperFluffy

consensus/src/marshal/mod.rs

@@ -62,10 +62,71 @@ pub mod ingress;
 pub use ingress::mailbox::Mailbox;
 pub mod resolver;
 
-use crate::{simplex::signing_scheme::Scheme, types::Epoch, Block};
-use futures::channel::oneshot;
+use crate::{simplex::signing_scheme::Scheme, types::Epoch, Block, Reporter};
+use futures::{
+    channel::oneshot,
+    future::{self, join, Either},
+};
 use std::sync::Arc;
 
+#[derive(Clone)]
+pub struct Reporters<L, R> {
+    left: L,
+    right: R,
+}
+
+impl<L, R> From<(L, R)> for Reporters<L, R>
+where
+    L: Reporter,
+    R: Reporter,
+{
+    fn from((left, right): (L, R)) -> Self {
+        Self { left, right }
+    }
+}
+
+impl<B, L, R> Reporter for Reporters<L, R>
+where
+    B: Block,
+    L: Reporter<Activity = Update<B>>,
+    R: Reporter<Activity = Update<B>>,
+{
+    type Activity = Update<B>;
+
+    async fn report(&mut self, activity: Self::Activity) {
+        let (to_left, to_right, to_await) = match activity {
+            Update::Tip(height, digest) => {
+                let to_left = Update::Tip(height, digest.clone());
+                let to_right = Update::Tip(height, digest.clone());
+                let to_await = Either::Left(future::ready(()));
+                (to_left, to_right, to_await)
+            }
+            Update::Block(block, report_ack) => {
+                let (to_left, left_ack) = {
+                    let (tx, rx) = oneshot::channel();
+                    let msg = Update::Block(block.clone(), tx);
+                    (msg, rx)
+                };
+                let (to_right, right_ack) = {
+                    let (tx, rx) = oneshot::channel();
+                    let msg = Update::Block(block.clone(), tx);
+                    (msg, rx)
+                };
+                let to_await = Either::Right(async move {
+                    if let Ok(((), ())) = future::try_join(left_ack, right_ack).await {
+                        let _ = report_ack.send(());
+                    } else {
+                        drop(report_ack);
+                    }
+                });
+                (to_left, to_right, to_await)
+            }
+        };
+        join(self.left.report(to_left), self.right.report(to_right)).await;
+        to_await.await
+    }
+}
+
 /// Supplies the signing scheme the marshal should use for a given epoch.
 pub trait SchemeProvider: Clone + Send + Sync + 'static {
     /// The signing scheme to provide.