Implementing Peer-to-Peer Direct Messaging

[MehranMazhar]

Hello libp2p community,
I'm currently working on a Rust-based project where I need to implement direct peer-to-peer messaging using the libp2p library.
My project already uses the libp2p gossipsub and mDNS protocols for broadcasting messages and discovering peers, respectively.
Now, I need to extend it to support direct messaging between peers.

Here's a simplified version of my current implementation:

use crate::node::blockchain::Blockchain;
use libp2p::{
    gossipsub, mdns, noise, swarm::Swarm, tcp, yamux, Multiaddr, PeerId,
    NetworkBehaviour, SwarmEvent,
};
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;

#[derive(NetworkBehaviour)]
pub struct P2PBehaviour {
    pub gossipsub: gossipsub::Behaviour,
    pub mdns: mdns::tokio::Behaviour,
}

pub struct P2PServer {
    pub behaviour: Swarm<P2PBehaviour>,
    pub topic: gossipsub::IdentTopic,
}

impl P2PServer {
    pub fn new(topic_name: &str, listen_addrs: &[&str], peer_addrs: &[&str]) -> Result<Self, Box<dyn std::error::Error>> {
        let mut swarm = Self::build_swarm(listen_addrs)?;
        let topic = Self::setup_gossipsub_topic(&mut swarm, topic_name)?;

        for peer in peer_addrs {
            let addr: Multiaddr = peer.parse()?;
            Swarm::dial(&mut swarm, addr)?;
        }

        Ok(Self { behaviour: swarm, topic })
    }

    async fn run(&mut self, blockchain: Arc<Mutex<Blockchain>>, mut command_rx: tokio::sync::mpsc::Receiver<P2PServerCommand>) -> Result<(), Box<dyn std::error::Error>> {
        self.process_messages(blockchain, &mut command_rx).await
    }

    async fn process_messages(&mut self, blockchain: Arc<Mutex<Blockchain>>, command_rx: &mut tokio::sync::mpsc::Receiver<P2PServerCommand>) -> Result<(), Box<dyn std::error::Error>> {
        loop {
            select! {
                event = self.behaviour.select_next_some().fuse() => {
                    Self::handle_swarm_event(event, &mut self.behaviour, &blockchain).await;
                },
                command = command_rx.recv() => {
                    if let Some(command) = command {
                        match command {
                            P2PServerCommand::SendGossipMessage { message, response_tx } => {
                                let result = self.send_gossip_message(message);
                                let _ = response_tx.send(result);
                            },
                            P2PServerCommand::GetConnectedPeers { response_tx } => {
                                let peers = self.get_connected_peers();
                                let _ = response_tx.send(peers);
                            }
                        }
                    }
                },
            }
        }
    }

    fn send_gossip_message(&mut self, message: Vec<u8>) -> Result<gossipsub::MessageId, gossipsub::PublishError> {
        self.behaviour.behaviour_mut().gossipsub.publish(self.topic.clone(), message)
    }

    fn get_connected_peers(&self) -> HashSet<PeerId> {
        self.behaviour.connected_peers().cloned().collect()
    }

    async fn handle_swarm_event(event: SwarmEvent<P2PBehaviourEvent>, swarm: &mut Swarm<P2PBehaviour>, blockchain: &Arc<Mutex<Blockchain>>) {
        // handle different swarm events here
    }
}

pub enum P2PServerCommand {
    SendGossipMessage {
        message: Vec<u8>,
        response_tx: tokio::sync::oneshot::Sender<Result<gossipsub::MessageId, gossipsub::PublishError>>,
    },
    GetConnectedPeers {
        response_tx: tokio::sync::oneshot::Sender<HashSet<PeerId>>,
    },
}

What I've Implemented:

• Gossipsub for broadcasting messages across the network.
• mDNS for peer discovery.
• Handling incoming swarm events and sending gossip messages.

What I Need Help With:
I want to extend this implementation to support direct messaging between peers using a request-response pattern. I understand that the
libp2p-request-response module is suitable for this, but I'm struggling with:

• Integrating the RequestResponse behavior into my existing P2PBehaviour.
• Handling direct message sending and receiving between specific peers.
• Best practices for efficiently managing connections and responses in a peer-to-peer setup.
• If anyone has experience or advice on implementing direct messaging with libp2p, I would greatly appreciate your insights. Sample code snippets or references to similar implementations would be incredibly helpful.

Thank you for your time and assistance!

[drHuangMHT]

Hi, I am currently developing an application with similar functionality, but there are differences in implementation details. You can check out my project here with a GUI here all on GitHub.
I personally haven't used request-response yet because I write a dedicated protocol for direct messaging. Though from what I can tell, the behavior needs two generic parameters for incoming requests and outgoing responses. If you are familiar with generics you shouldn't have any problem integrating the behavior(either cbor::Behaviour or json::Behaviour) into your existing behaviors.
The message to send should be contained in the request, with response being the receipt of the message. For the receiver, listen on Event::Message for incoming messages.
As for connections, libp2p has built-in timeout and keep-alive mechanism. As long as you are not negotiating infinite amount of substreams(with your custom behavior), you should be fine. Typically, only one "physical" connection(TCP connection for example) will be established for every peer that carries multiple streams dedicated to different ConnectionHandlers, and on one stream the handler can request for multiple substreams.