use type alias for network adapter (#256)

nomos-services/consensus/src/network/adapters/mock.rs

@@ -6,12 +6,12 @@ use nomos_network::{
     NetworkMsg, NetworkService,
 };
 use overwatch_rs::services::{relay::OutboundRelay, ServiceData};
-use tokio_stream::{wrappers::BroadcastStream, Stream};
+use tokio_stream::wrappers::BroadcastStream;
 
 use crate::network::messages::{NewViewMsg, TimeoutMsg, TimeoutQcMsg};
 use crate::network::{
     messages::{ProposalChunkMsg, VoteMsg},
-    NetworkAdapter,
+    BoxedStream, NetworkAdapter,
 };
 use consensus_engine::{BlockId, Committee, View};
 
@@ -57,10 +57,7 @@ impl NetworkAdapter for MockAdapter {
         Self { network_relay }
     }
 
-    async fn proposal_chunks_stream(
+    async fn proposal_chunks_stream(&self, _view: View) -> BoxedStream<ProposalChunkMsg> {
-        &self,
-        _view: View,
-    ) -> Box<dyn Stream<Item = ProposalChunkMsg> + Send + Sync + Unpin> {
         let stream_channel = self
             .message_subscriber_channel()
             .await
@@ -110,27 +107,15 @@ impl NetworkAdapter for MockAdapter {
         todo!()
     }
 
-    async fn timeout_stream(
+    async fn timeout_stream(&self, _committee: &Committee, _view: View) -> BoxedStream<TimeoutMsg> {
-        &self,
-        _committee: &Committee,
-        _view: View,
-    ) -> Box<dyn Stream<Item = TimeoutMsg> + Send + Sync + Unpin> {
         todo!()
     }
 
-    async fn timeout_qc_stream(
+    async fn timeout_qc_stream(&self, _view: View) -> BoxedStream<TimeoutQcMsg> {
-        &self,
-        _view: View,
-    ) -> Box<dyn Stream<Item = TimeoutQcMsg> + Send + Sync + Unpin> {
         todo!()
     }
 
-    async fn votes_stream(
+    async fn votes_stream(&self, _: &Committee, _: View, _: BlockId) -> BoxedStream<VoteMsg> {
-        &self,
-        _committee: &Committee,
-        _view: View,
-        _proposal_id: BlockId,
-    ) -> Box<dyn Stream<Item = VoteMsg> + Send + Unpin> {
         let stream_channel = self
             .message_subscriber_channel()
             .await
@@ -156,11 +141,7 @@ impl NetworkAdapter for MockAdapter {
         )))
     }
 
-    async fn new_view_stream(
+    async fn new_view_stream(&self, _: &Committee, _view: View) -> BoxedStream<NewViewMsg> {
-        &self,
-        _committee: &Committee,
-        _view: View,
-    ) -> Box<dyn Stream<Item = NewViewMsg> + Send + Unpin> {
         todo!()
     }
 

nomos-services/consensus/src/network/adapters/waku.rs

@@ -10,7 +10,7 @@ use tokio_stream::wrappers::BroadcastStream;
 use crate::network::messages::{NewViewMsg, TimeoutMsg, TimeoutQcMsg};
 use crate::network::{
     messages::{ProposalChunkMsg, VoteMsg},
-    NetworkAdapter,
+    BoxedStream, NetworkAdapter,
 };
 use consensus_engine::{BlockId, Committee, View};
 use nomos_network::{
@@ -57,10 +57,7 @@ impl WakuAdapter {
     async fn archive_subscriber_stream(
         network_relay: OutboundRelay<<NetworkService<Waku> as ServiceData>::Message>,
         content_topic: WakuContentTopic,
-    ) -> Result<
+    ) -> Result<BoxedStream<WakuMessage>, tokio::sync::oneshot::error::RecvError> {
-        Box<dyn Stream<Item = WakuMessage> + Send + Sync + Unpin>,
-        tokio::sync::oneshot::error::RecvError,
-    > {
         let (sender, receiver) = tokio::sync::oneshot::channel();
         if let Err((_, _e)) = network_relay
             .send(NetworkMsg::Process(WakuBackendMessage::ArchiveSubscribe {
@@ -167,10 +164,7 @@ impl NetworkAdapter for WakuAdapter {
         Self { network_relay }
     }
 
-    async fn proposal_chunks_stream(
+    async fn proposal_chunks_stream(&self, view: View) -> BoxedStream<ProposalChunkMsg> {
-        &self,
-        view: View,
-    ) -> Box<dyn Stream<Item = ProposalChunkMsg> + Send + Sync + Unpin> {
         Box::new(Box::pin(
             self.cached_stream_with_content_topic(PROPOSAL_CONTENT_TOPIC)
                 .await
@@ -214,11 +208,7 @@ impl NetworkAdapter for WakuAdapter {
             .await
     }
 
-    async fn timeout_stream(
+    async fn timeout_stream(&self, committee: &Committee, view: View) -> BoxedStream<TimeoutMsg> {
-        &self,
-        committee: &Committee,
-        view: View,
-    ) -> Box<dyn Stream<Item = TimeoutMsg> + Send + Sync + Unpin> {
         let content_topic = create_topic("timeout", committee, view);
         Box::new(Box::pin(
             self.cached_stream_with_content_topic(content_topic)
@@ -237,10 +227,7 @@ impl NetworkAdapter for WakuAdapter {
         ))
     }
 
-    async fn timeout_qc_stream(
+    async fn timeout_qc_stream(&self, view: View) -> BoxedStream<TimeoutQcMsg> {
-        &self,
-        view: View,
-    ) -> Box<dyn Stream<Item = TimeoutQcMsg> + Send + Sync + Unpin> {
         Box::new(Box::pin(
             self.cached_stream_with_content_topic(TIMEOUT_QC_CONTENT_TOPIC)
                 .await
@@ -263,7 +250,7 @@ impl NetworkAdapter for WakuAdapter {
         committee: &Committee,
         view: View,
         proposal_id: BlockId,
-    ) -> Box<dyn Stream<Item = VoteMsg> + Send + Unpin> {
+    ) -> BoxedStream<VoteMsg> {
         let content_topic = create_topic("votes", committee, view);
         Box::new(Box::pin(
             self.cached_stream_with_content_topic(content_topic)
@@ -282,11 +269,7 @@ impl NetworkAdapter for WakuAdapter {
         ))
     }
 
-    async fn new_view_stream(
+    async fn new_view_stream(&self, committee: &Committee, view: View) -> BoxedStream<NewViewMsg> {
-        &self,
-        committee: &Committee,
-        view: View,
-    ) -> Box<dyn Stream<Item = NewViewMsg> + Send + Unpin> {
         let content_topic = create_topic("new-view", committee, view);
         Box::new(Box::pin(
             self.cached_stream_with_content_topic(content_topic)

nomos-services/consensus/src/network/mod.rs

@@ -12,6 +12,8 @@ use nomos_network::NetworkService;
 use overwatch_rs::services::relay::OutboundRelay;
 use overwatch_rs::services::ServiceData;
 
+type BoxedStream<T> = Box<dyn Stream<Item = T> + Send + Sync + Unpin>;
+
 #[async_trait::async_trait]
 pub trait NetworkAdapter {
     type Backend: NetworkBackend + 'static;
@@ -24,25 +26,14 @@ pub trait NetworkAdapter {
     ) -> Box<dyn Stream<Item = ProposalChunkMsg> + Send + Sync + Unpin>;
     async fn broadcast_block_chunk(&self, chunk_msg: ProposalChunkMsg);
     async fn broadcast_timeout_qc(&self, timeout_qc_msg: TimeoutQcMsg);
-    async fn timeout_stream(
+    async fn timeout_stream(&self, committee: &Committee, view: View) -> BoxedStream<TimeoutMsg>;
-        &self,
+    async fn timeout_qc_stream(&self, view: View) -> BoxedStream<TimeoutQcMsg>;
-        committee: &Committee,
-        view: View,
-    ) -> Box<dyn Stream<Item = TimeoutMsg> + Send + Sync + Unpin>;
-    async fn timeout_qc_stream(
-        &self,
-        view: View,
-    ) -> Box<dyn Stream<Item = TimeoutQcMsg> + Send + Sync + Unpin>;
     async fn votes_stream(
         &self,
         committee: &Committee,
         view: View,
         proposal_id: BlockId,
-    ) -> Box<dyn Stream<Item = VoteMsg> + Send + Unpin>;
+    ) -> BoxedStream<VoteMsg>;
-    async fn new_view_stream(
+    async fn new_view_stream(&self, committee: &Committee, view: View) -> BoxedStream<NewViewMsg>;
-        &self,
-        committee: &Committee,
-        view: View,
-    ) -> Box<dyn Stream<Item = NewViewMsg> + Send + Unpin>;
     async fn send(&self, committee: &Committee, view: View, payload: Box<[u8]>, channel: &str);
 }