make receiver inactive

EspressoSystems · Apr 10, 2024 · b9b7702 · b9b7702
1 parent ddb375d
commit b9b7702
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 6 deletions.
diff --git a/src/events_source.rs b/src/events_source.rs
@@ -1,4 +1,4 @@
-use async_broadcast::{broadcast, Receiver as BroadcastReceiver, Sender as BroadcastSender};
+use async_broadcast::{broadcast, InactiveReceiver, Sender as BroadcastSender};
 use async_trait::async_trait;
 use futures::future::BoxFuture;
 use futures::stream::{self, BoxStream, Stream, StreamExt};
@@ -123,7 +123,7 @@ where
 #[derive(Debug)]
 pub struct EventsStreamer<Types: NodeType> {
     // required for api subscription
-    to_subscribe_clone_recv: BroadcastReceiver<Arc<BuilderEvent<Types>>>,
+    inactive_to_subscribe_clone_recv: InactiveReceiver<Arc<BuilderEvent<Types>>>,
     subscriber_send_channel: BroadcastSender<Arc<BuilderEvent<Types>>>,
 
     // required for sending startup info
@@ -165,7 +165,7 @@ impl<Types: NodeType> EventsSource<Types> for EventsStreamer<Types> {
     type EventStream = BoxStream<'static, Arc<BuilderEvent<Types>>>;
 
     async fn get_event_stream(&self) -> Self::EventStream {
-        let recv_channel = self.to_subscribe_clone_recv.clone();
+        let recv_channel = self.inactive_to_subscribe_clone_recv.activate_cloned();
         let starup_event_initialized = false;
         let startup_event = self.get_startup_event().clone();
         stream::unfold(
@@ -196,9 +196,10 @@ impl<Types: NodeType> EventsStreamer<Types> {
         let (mut subscriber_send_channel, to_subscribe_clone_recv) =
             broadcast::<Arc<BuilderEvent<Types>>>(RETAINED_EVENTS_COUNT);
         subscriber_send_channel.set_overflow(true);
+        let inactive_to_subscribe_clone_recv = to_subscribe_clone_recv.deactivate();
         EventsStreamer {
             subscriber_send_channel,
-            to_subscribe_clone_recv,
+            inactive_to_subscribe_clone_recv,
             known_nodes_with_stake,
             non_staked_node_count,
         }

diff --git a/src/test.rs b/src/test.rs
@@ -70,7 +70,7 @@ mod tests {
 
         tracing::info!("Client 1 Connected to server");
 
-        // client subscrive to hotshot events
+        // client 1 subscribe to hotshot events
         let mut events_1 = client_1
             .socket("events")
             .subscribe::<BuilderEvent<TestTypes>>()
@@ -89,7 +89,7 @@ mod tests {
 
         tracing::info!("Client 2 Connected to server");
 
-        // client 1  subscrive to hotshot events
+        // client 2 subscrive to hotshot events
         let mut events_2 = client_2
             .socket("events")
             .subscribe::<BuilderEvent<TestTypes>>()