fn deliver_message(
        &mut self,
        peer: &PeerId,
        num_messages: usize,
    ) -> (Vec<SignedTransaction>, PeerId) {
        // await broadcast notification
        for _ in 0..num_messages {
            self.wait_for_event(peer, SharedMempoolNotification::Broadcast);
        }

        // await next message from node
        let network_reqs_rx = self.network_reqs_rxs.get_mut(peer).unwrap();
        let network_req = block_on(network_reqs_rx.next()).unwrap();

        if let PeerManagerRequest::SendMessage(peer_id, msg) = network_req {
            let sync_msg = lcs::from_bytes(&msg.mdata).unwrap();
            if let MempoolSyncMsg::BroadcastTransactionsRequest { transactions, .. } = sync_msg {
                // send it to peer
                let receiver_network_notif_tx = self.network_notifs_txs.get_mut(&peer_id).unwrap();
                receiver_network_notif_tx
                    .push(
                        (*peer, ProtocolId::MempoolDirectSend),
                        PeerManagerNotification::RecvMessage(*peer, msg),
                    )
                    .unwrap();

                // await message delivery
                self.wait_for_event(&peer_id, SharedMempoolNotification::NewTransactions);

                // verify transaction was inserted into Mempool
                let mempool = self.mempools.get(&peer_id).unwrap();
                let block = mempool.lock().unwrap().get_block(100, HashSet::new());
                for txn in transactions.iter() {
                    assert!(block.contains(txn));
                }

                // deliver ACK for this request
                self.deliver_response(&peer_id);
                (transactions, peer_id)
            } else {
                panic!("did not receive expected BroadcastTransactionsRequest");
            }
        } else {
            panic!("peer {:?} didn't broadcast transaction", peer)
        }
    }
/// delivers next broadcast message from `peer`
delivers next broadcast message from `peer`
delivers next broadcast message from ` peer `
fn deliver_message(
        &mut self,
        peer: &PeerId,
        num_messages: usize,
    ) -> (Vec<SignedTransaction>, PeerId) {
        // await broadcast notification
        for _ in 0..num_messages {
            self.wait_for_event(peer, SharedMempoolNotification::Broadcast);
        }
        // await next message from node
        let network_reqs_rx = self.network_reqs_rxs.get_mut(peer).unwrap();
        let network_req = block_on(network_reqs_rx.next()).unwrap();
        if let PeerManagerRequest::SendMessage(peer_id, msg) = network_req {
            let sync_msg = lcs::from_bytes(&msg.mdata).unwrap();
            if let MempoolSyncMsg::BroadcastTransactionsRequest { transactions, .. } = sync_msg {
                // send it to peer
                let receiver_network_notif_tx = self.network_notifs_txs.get_mut(&peer_id).unwrap();
                receiver_network_notif_tx
                    .push(
                        (*peer, ProtocolId::MempoolDirectSend),
                        PeerManagerNotification::RecvMessage(*peer, msg),
                    )
                    .unwrap();
                // await message delivery
                self.wait_for_event(&peer_id, SharedMempoolNotification::NewTransactions);
                // verify transaction was inserted into Mempool
                let mempool = self.mempools.get(&peer_id).unwrap();
                let block = mempool.lock().unwrap().get_block(100, HashSet::new());
                for txn in transactions.iter() {
                    assert!(block.contains(txn));
                }
                // deliver ACK for this request
                self.deliver_response(&peer_id);
                (transactions, peer_id)
            } else {
                panic!("did not receive expected BroadcastTransactionsRequest");
            }
        } else {
            panic!("peer {:?} didn't broadcast transaction", peer)
        }
    }
fn deliver_message ( & mut self , peer : & PeerId , num_messages : usize , ) -> ( Vec < SignedTransaction > , PeerId ) { // await broadcast notification for _ in 0 .. num_messages { self . wait_for_event ( peer , SharedMempoolNotification :: Broadcast ) ; } // await next message from node let network_reqs_rx = self . network_reqs_rxs . get_mut ( peer ) . unwrap ( ) ; let network_req = block_on ( network_reqs_rx . next ( ) ) . unwrap ( ) ; if let PeerManagerRequest :: SendMessage ( peer_id , msg ) = network_req { let sync_msg = lcs :: from_bytes ( & msg . mdata ) . unwrap ( ) ; if let MempoolSyncMsg :: BroadcastTransactionsRequest { transactions , .. } = sync_msg { // send it to peer let receiver_network_notif_tx = self . network_notifs_txs . get_mut ( & peer_id ) . unwrap ( ) ; receiver_network_notif_tx . push ( ( * peer , ProtocolId :: MempoolDirectSend ) , PeerManagerNotification :: RecvMessage ( * peer , msg ) , ) . unwrap ( ) ; // await message delivery self . wait_for_event ( & peer_id , SharedMempoolNotification :: NewTransactions ) ; // verify transaction was inserted into Mempool let mempool = self . mempools . get ( & peer_id ) . unwrap ( ) ; let block = mempool . lock ( ) . unwrap ( ) . get_block ( 100 , HashSet :: new ( ) ) ; for txn in transactions . iter ( ) { assert ! ( block contains ( txn ) ) ; } // deliver ACK for this request self . deliver_response ( & peer_id ) ; ( transactions , peer_id ) } else { panic ! ( " " ) ; } } else { panic ! ( " " peer ) } }
delivers next broadcast message from `peer`
delivers next broadcast message from ` peer `
&mut self
peer
&PeerId
num_messages
usize
&mut self
peer
&PeerId
num_messages
usize