[broadcast/buffer] Final Tweaks (#837)

Author: patrick-ogrady

broadcast/src/buffered/engine.rs

@@ -32,8 +32,6 @@ pub struct Engine<
     D: Digest,
     Cfg: CodecCfg,
     M: Digestible<D> + Codec<Cfg>,
-    NetS: Sender<PublicKey = P>,
-    NetR: Receiver<PublicKey = P>,
 > {
     ////////////////////////////////////////
     // Interfaces
@@ -87,8 +85,6 @@ pub struct Engine<
     ////////////////////////////////////////
     /// Metrics
     metrics: metrics::Metrics,
-
-    _phantom: std::marker::PhantomData<(NetS, NetR)>,
 }
 
 impl<
@@ -97,9 +93,7 @@ impl<
         D: Digest,
         Cfg: CodecCfg,
         M: Digestible<D> + Codec<Cfg>,
-        NetS: Sender<PublicKey = P>,
-        NetR: Receiver<PublicKey = P>,
-    > Engine<E, P, D, Cfg, M, NetS, NetR>
+    > Engine<E, P, D, Cfg, M>
 {
     /// Creates a new engine with the given context and configuration.
     /// Returns the engine and a mailbox for sending messages to the engine.
@@ -120,20 +114,21 @@ impl<
             items: HashMap::new(),
             counts: HashMap::new(),
             metrics,
-
-            _phantom: std::marker::PhantomData,
         };
 
         (result, mailbox)
     }
 
     /// Starts the engine with the given network.
-    pub fn start(mut self, network: (NetS, NetR)) -> Handle<()> {
+    pub fn start(
+        mut self,
+        network: (impl Sender<PublicKey = P>, impl Receiver<PublicKey = P>),
+    ) -> Handle<()> {
         self.context.spawn_ref()(self.run(network))
     }
 
     /// Inner run loop called by `start`.
-    async fn run(mut self, network: (NetS, NetR)) {
+    async fn run(mut self, network: (impl Sender<PublicKey = P>, impl Receiver<PublicKey = P>)) {
         let (mut sender, mut receiver) = wrap(self.codec_config.clone(), network.0, network.1);
         let mut shutdown = self.context.stopped();
 
@@ -204,9 +199,9 @@ impl<
     ////////////////////////////////////////
 
     /// Handles a `broadcast` request from the application.
-    async fn handle_broadcast(
+    async fn handle_broadcast<Sr: Sender<PublicKey = P>>(
         &mut self,
-        sender: &mut WrappedSender<NetS, Cfg, M>,
+        sender: &mut WrappedSender<Sr, Cfg, M>,
         msg: M,
         responder: oneshot::Sender<Vec<P>>,
     ) {

broadcast/src/buffered/mod.rs

@@ -121,10 +121,7 @@ mod tests {
                 codec_config: (),
             };
             let (engine, engine_mailbox) =
-                Engine::<_, PublicKey, Sha256Digest, _, TestMessage, _, _>::new(
-                    context.clone(),
-                    config,
-                );
+                Engine::<_, PublicKey, Sha256Digest, _, TestMessage>::new(context.clone(), config);
             mailboxes.insert(peer.clone(), engine_mailbox);
             engine.start(network);
         }
@@ -156,6 +153,27 @@ mod tests {
                 assert_eq!(received_message.unwrap(), message);
             }
             assert_eq!(result.await.unwrap().len(), peers.len() - 1);
+
+            // Drop broadcast result
+            let message = TestMessage::new(b"hello world again");
+            let result = first_mailbox.broadcast(message.clone()).await;
+            drop(result);
+
+            // Allow time for propagation
+            context.sleep(Duration::from_secs(1)).await;
+
+            // Check that all peers received the new message
+            let mut found = 0;
+            for peer in peers.iter() {
+                let mut mailbox = mailboxes.get(peer).unwrap().clone();
+                let digest = message.digest();
+                let receiver = mailbox.get(digest).await;
+                if let Some(msg) = receiver {
+                    assert_eq!(msg, message);
+                    found += 1;
+                }
+            }
+            assert!(found > 0, "No peers received the message");
         });
     }