add msgs to msg pool

blockchain/message_pool/src/block_prob.rs

@@ -16,15 +16,13 @@ fn poiss_pdf(x: f64, mu: f64, cond: f64) -> f64 {
 
 pub fn no_winners_prob() -> Vec<f64> {
     (0..MAX_BLOCKS)
-        .into_iter()
         .map(|i| poiss_pdf(i as f64, MU, MU))
         .collect()
 }
 
 fn no_winners_prob_assuming_more_than_one() -> Vec<f64> {
     let cond = (E.powf(5.0) - 1.0).log(E);
     (0..MAX_BLOCKS)
-        .into_iter()
         .map(|i| poiss_pdf(i as f64, MU, cond))
         .collect()
 }
@@ -76,7 +74,6 @@ pub fn block_probabilities(tq: f64) -> Vec<f64> {
     let no_winners = no_winners_prob_assuming_more_than_one();
     let p = 1.0 - tq;
     (0..MAX_BLOCKS)
-        .into_iter()
         .map(|place| {
             no_winners
                 .iter()

blockchain/message_pool/src/msgpool.rs

@@ -225,7 +225,7 @@ where
 }
 
 /// This is the main MessagePool struct
-pub struct MessagePool<T: 'static> {
+pub struct MessagePool<T> {
     local_addrs: Arc<RwLock<Vec<Address>>>,
     pending: Arc<RwLock<HashMap<Address, MsgSet>>>,
     pub cur_tipset: Arc<RwLock<Tipset>>,

forest/src/daemon.rs

@@ -121,12 +121,6 @@ pub(super) async fn start(config: Config) {
         .await
         .unwrap();
 
-    // Libp2p service setup
-    let p2p_service =
-        Libp2pService::new(config.network, Arc::clone(&db), net_keypair, &network_name);
-    let network_rx = p2p_service.network_receiver();
-    let network_send = p2p_service.network_sender();
-
     // Initialize mpool
     let publisher = chain_store.publisher();
     let subscriber = publisher.write().await.subscribe();
@@ -141,6 +135,17 @@ pub(super) async fn start(config: Config) {
         .unwrap(),
     );
 
+    // Libp2p service setup
+    let p2p_service = Libp2pService::new(
+        config.network,
+        Arc::clone(&db),
+        Arc::clone(&mpool),
+        net_keypair,
+        &network_name,
+    );
+    let network_rx = p2p_service.network_receiver();
+    let network_send = p2p_service.network_sender();
+
     // Get Drand Coefficients
     let coeff = config.drand_public;
 

node/forest_libp2p/Cargo.toml

@@ -41,6 +41,7 @@ libp2p-bitswap = "=0.6.1"
 tiny-cid = "0.2.0"
 ipld_blockstore = { path = "../../ipld/blockstore" }
 async-trait = "0.1"
+message_pool = { path = "../../blockchain/message_pool" }
 
 [dev-dependencies]
 forest_address = { path = "../../vm/address" }

node/forest_libp2p/src/service.rs

@@ -7,12 +7,15 @@ use super::{ForestBehaviour, ForestBehaviourEvent, Libp2pConfig};
 use crate::hello::{HelloRequest, HelloResponse};
 use async_std::sync::{channel, Receiver, Sender};
 use async_std::{stream, task};
+use forest_blocks::GossipBlock;
 use forest_cid::{multihash::Blake2b256, Cid};
 use forest_encoding::from_slice;
+use forest_message::SignedMessage;
 use futures::channel::oneshot::Sender as OneShotSender;
 use futures::select;
 use futures_util::stream::StreamExt;
 use ipld_blockstore::BlockStore;
+pub use libp2p::gossipsub::Topic;
 use libp2p::{
     core,
     core::muxing::StreamMuxerBox,
@@ -22,16 +25,13 @@ use libp2p::{
 };
 use libp2p_request_response::{RequestId, ResponseChannel};
 use log::{debug, error, info, trace, warn};
+use message_pool::{MessagePool, Provider};
 use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 use std::time::Duration;
 use utils::read_file_to_vec;
 
-use forest_blocks::GossipBlock;
-use forest_message::SignedMessage;
-pub use libp2p::gossipsub::Topic;
-
 pub const PUBSUB_BLOCK_STR: &str = "/fil/blocks";
 pub const PUBSUB_MSG_STR: &str = "/fil/msgs";
 
@@ -95,9 +95,10 @@ pub enum NetworkMessage {
     },
 }
 /// The Libp2pService listens to events from the Libp2p swarm.
-pub struct Libp2pService<DB: BlockStore> {
+pub struct Libp2pService<DB, T> {
     pub swarm: Swarm<ForestBehaviour>,
     db: Arc<DB>,
+    mpool: Arc<MessagePool<T>>,
     /// Keeps track of Blocksync requests to responses
     bs_request_table: HashMap<RequestId, OneShotSender<BlockSyncResponse>>,
     network_receiver_in: Receiver<NetworkMessage>,
@@ -107,14 +108,16 @@ pub struct Libp2pService<DB: BlockStore> {
     network_name: String,
 }
 
-impl<DB> Libp2pService<DB>
+impl<DB, T> Libp2pService<DB, T>
 where
     DB: BlockStore + Sync + Send + 'static,
+    T: Provider + Sync + Send + 'static,
 {
     /// Constructs a Libp2pService
     pub fn new(
         config: Libp2pConfig,
         db: Arc<DB>,
+        mpool: Arc<MessagePool<T>>,
         net_keypair: Keypair,
         network_name: &str,
     ) -> Self {
@@ -146,6 +149,7 @@ where
         Libp2pService {
             swarm,
             db,
+            mpool,
             bs_request_table: HashMap::new(),
             network_receiver_in,
             network_sender_in,
@@ -205,8 +209,12 @@ where
                                     Ok(m) => {
                                         emit_event(&self.network_sender_out, NetworkEvent::PubsubMessage{
                                             source,
-                                            message: PubsubMessage::Message(m),
+                                            message: PubsubMessage::Message(m.clone()),
                                         }).await;
+                                        // add message to message pool
+                                        if let Err(e) = self.mpool.add(m).await {
+                                            warn!("Gossip Message failed to be added to Message pool");
+                                        }
                                     }
                                     Err(e) => warn!("Gossip Message from peer {:?} could not be deserialized: {}", source, e)
                                 }