REFACTOR: try_from to filter_gossipsub_messages, Vec<msg_handlers> to…

… msg_handler

src/conversions.rs

@@ -9,7 +9,7 @@
 //!     - From<[PublicAddress]> for [ParallelChain PublicAddress](pchain_types::cryptography::PublicAddress)
 //!     - TryFrom<PeerId> for [PublicAddress]
 //!     - TryFrom<[PublicAddress]> for [PeerId]
-//!     - TryFrom<([libp2p::gossipsub::Message], [pchain_types::cryptography::PublicAddress])> for [Message]
+//!     - filter_gossipsub_messages([libp2p::gossipsub::Message], [pchain_types::cryptography::PublicAddress]) to [Message]
 
 use libp2p::identity::{self, ed25519, DecodingError, OtherVariantError, PeerId};
 use libp2p::{Multiaddr, kad::KBucketKey};
@@ -77,13 +77,11 @@ impl From<DecodingError> for PublicAddressTryFromPeerIdError {
     }
 }
 
-impl TryFrom<(libp2p::gossipsub::Message, cryptography::PublicAddress)> for Message {
-    type Error = MessageConversionError;
-
-    fn try_from((message , local_public_address): (libp2p::gossipsub::Message, cryptography::PublicAddress)) 
-    -> Result<Self, Self::Error> {
-        let (topic_hash, data) = (message.topic, message.data);
-        let mut data = data.as_slice();
+/// converts [Message](libp2p::gossipsub::Message) to [Message](Message) while 
+/// filtering for message topics that can be forwarded to fullnode
+pub fn filter_gossipsub_messages(message: libp2p::gossipsub::Message, local_public_address: cryptography::PublicAddress) -> Result<Message, MessageConversionError> {
+    let (topic_hash, data) = (message.topic, message.data);
+    let mut data = data.as_slice();
 
         let topic = fullnode_topics(local_public_address)
             .into_iter()
@@ -100,7 +98,6 @@ impl TryFrom<(libp2p::gossipsub::Message, cryptography::PublicAddress)> for Mess
                 Ok(message)
             },
         }
-    }
 }
 
 #[derive(Debug)]

src/lib.rs

@@ -26,18 +26,18 @@
 //! };
 //!
 //! // 2. Create message handlers 
-//! let (tx, rx) = mpsc::channel();
+//! let (tx, rx) =  tokio::sync::mpsc::channel(100);
 //! let hotstuff_sender = tx.clone();
 //! let hotstuff_handler = move |msg_orgin: [u8;32], msg: Message| {
 //!     match msg {
 //!         Message::HotStuffRs(hotstuff_message) => {
 //!             //process hotstuff message
-//!             let _ = hotstuff_sender.send((msg_origin, Message::HotStuffRs(hotstuff_message)));
+//!             let _ = hotstuff_sender.try_send((msg_origin, Message::HotStuffRs(hotstuff_message)));
 //!         }
 //!         _ => {}
 //!     }  
 //! };
-//! let mut message_handlers: Vec<Box<dyn Fn(PublicAddress, Message) + Send>> = vec![];
+//! let mut message_handlers: Vec<Box<dyn FnMut(PublicAddress, Message) + Send>> = vec![];
 //! message_handlers.push(Box::new(hotstuff_handler));
 //!  
 //! // 3. Start P2P network.
@@ -47,7 +47,7 @@
 //! 
 //! // 4. Send out messages.
 //! peer.broadcast_mempool_msg(txn);
-//!
+//! ```
 //!
 
 

src/peer.rs

@@ -14,25 +14,29 @@
 //! Starting Peer will return a Sender for delivering PeerCommand to the thread. 
 //! 
 //! Example:
-//!
-//! // 1. Define the configurations
+//! 
+//! 1. Define the configurations
+//! ```
 //! let config = Config {...}
-//!
-//! // 2. Define the message handlers 
+//! ```
+//! 2. Define the message handlers 
+//! ```
 //! let(tx,rx) = mpsc::channel();
 //! let message_sender = tx.clone();
 //! let message_handler = move |msg_origin: [u8;32], msg: Message| {
 //!     let _ = message_sender.send((msg_origin, msg));
 //! };
 //! let mut message_handlers: Vec<Box<dyn Fn(PublicAddress, Message) + Send>> = vec![];
 //! message_handlers.push(Box::new(message_handler));
-//! 
-//! // 3. Start the peer
+//!  ```
+//! 3. Start the peer
+//!  ```
 //! let peer = Peer::start(config, message_handlers).await.unwrap();
-//! 
-//! // 4. Send PeerCommand
+//!  ```
+//! 4. Send PeerCommand
+//!  ```
 //! peer.broadcast_mempool_msg(txn);
-//! 
+//!  ```
 
 use futures::StreamExt;
 use tokio::task::JoinHandle;
@@ -72,7 +76,7 @@ impl Peer {
 /// 3. Establishes connection to the network by adding bootnodes and subscribing to message [Topic]. 
 /// 4. Spawns an asynchronous [tokio] task and enters the event handling loop, returning a Sender used for sending 
 /// [PeerCommand] to the internal thread.
-    pub async fn start(config: Config, handlers: Vec<Box<dyn FnMut(PublicAddress, Message) + Send>>) -> Result<Peer, PeerStartError> {
+    pub async fn start(config: Config, handler: Box<dyn FnMut(PublicAddress, Message) + Send>) -> Result<Peer, PeerStartError> {
 
         let mut swarm = set_up_transport(&config)
         .await
@@ -87,7 +91,7 @@ impl Peer {
         swarm = establish_network_connections(swarm, &config)
         .map_err(PeerStartError::SubscriptionError)?;
 
-        let (handle, sender) = start_event_handling(swarm, &config, handlers);
+        let (handle, sender) = start_event_handling(swarm, &config, handler);
         Ok(
             Peer {
                 handle,
@@ -215,7 +219,7 @@ fn establish_network_connections(mut swarm: libp2p::Swarm<Behaviour> , config: &
 /// Upon receiving a (Shutdown)[PeerCommand::Shutdown] command, the process will exit the loop and terminate
 /// the thread.
 /// 
-fn start_event_handling(mut swarm: libp2p::Swarm<Behaviour>, config: &Config, mut message_handlers: Vec<Box<dyn FnMut(PublicAddress, Message) + Send>>) -> 
+fn start_event_handling(mut swarm: libp2p::Swarm<Behaviour>, config: &Config, mut message_handler: Box<dyn FnMut(PublicAddress, Message) + Send>) -> 
     (JoinHandle<()>,tokio::sync::mpsc::Sender<PeerCommand>) {
     // 4. Start p2p networking
     let local_keypair = &config.keypair;
@@ -259,8 +263,7 @@ fn start_event_handling(mut swarm: libp2p::Swarm<Behaviour>, config: &Config, mu
                         log::info!("Publishing (Topic: {:?})", topic);
                         if swarm.behaviour().is_subscribed(&topic.clone().hash()) {
                             // Send it to ourselves if we subscribed to this topic
-                            message_handlers.iter_mut()
-                            .for_each(|handler| handler(local_public_address, message.clone()));
+                            message_handler(local_public_address, message.clone());
                         } 
                         if let Err(e) = swarm.behaviour_mut().publish(topic, message) {
                             log::debug!("Failed to publish the message. {:?}", e);
@@ -288,11 +291,9 @@ fn start_event_handling(mut swarm: libp2p::Swarm<Behaviour>, config: &Config, mu
                                 if swarm.behaviour().is_subscribed(&message.topic) {
                                     // Send it to ourselves if we subscribed to this topic
                                     if let Ok(message) =
-                                        Message::try_from((message, local_public_address))
+                                        conversions::filter_gossipsub_messages(message, local_public_address)
                                     {
-                                        message_handlers.iter_mut().for_each(|handler| {
-                                            handler(public_addr, message.clone())
-                                        });
+                                        message_handler(public_addr, message.clone())
                                     }                                  
                                 }
                             }

tests/network.rs

@@ -486,33 +486,21 @@ pub async fn node(
 
     let(tx,rx) = mpsc::channel(100);
 
-    let hotstuff_sender = tx.clone();
-    let hotstuff_handler = move |msg_origin: [u8;32], msg: Message| {
+    let message_sender = tx.clone();
+    let message_handler = move |msg_origin: [u8;32], msg: Message| {
         match msg {
             Message::HotStuffRs(hotstuff_message) => {
                 // process hotstuff message
-                let _ = hotstuff_sender.try_send((msg_origin, Message::HotStuffRs(hotstuff_message)));
+                let _ = message_sender.try_send((msg_origin, Message::HotStuffRs(hotstuff_message)));
             }
-            _ => {}
-        }
-    };
-
-    let message_sender = tx.clone();
-    let mempool_handler = move |msg_origin: [u8;32], msg: Message| {
-        match msg {
             Message::Mempool(mempool_message) => {
                 // process mempool message
                 let _ = message_sender.try_send((msg_origin, Message::Mempool(mempool_message)));
             }
-            _ => {}
         }
     };
 
-    let mut message_handlers: Vec<Box<dyn FnMut(PublicAddress, Message) + Send>> = vec![];
-    message_handlers.push(Box::new(hotstuff_handler));
-    message_handlers.push(Box::new(mempool_handler));
-
-    let peer = Peer::start(config, message_handlers).await.unwrap();
+    let peer = Peer::start(config, Box::new(message_handler)).await.unwrap();
 
     (peer, rx)
 }