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.

[Lebei2046]

Hi, all,
I tried to write a chat app , just finished mnemonic signup and adding friend by scanning qr-code. I'm also looking for p2p direct messaging solution.

I still have questions about libp2p.

• Is libp2p reliable to write chat apps for group messaging or p2p direct messaging?
• How deal with the situation that the peers are offline?

[dariusc93]

Is libp2p reliable to write chat apps for group messaging or p2p direct messaging?

Yes, though a lot of the logic comes down to how you implement it. Ideally you could use gossipsub but you could get by with using the oneshot transport or request-response for direct messaging, while group chat could use gossipsub with a unique topic per group. You would want to make sure the peers are apart of the mesh in some manner to receive the messages and likely want to deal with encryption.

How deal with the situation that the peers are offline?

In one of my personal projects (I havent made it public yet), when a peer comes online it would check multiple sources for updates, but you would want to have a separate node that is always online that can store messages for another peer thats apart of that chat. The other option is the synchronize the chat when both peers are online. I think in your case though it might be best to look into CRDT to assist.

[Lebei2046]

With the help of AI, I have written a lib gigi-p2p combined nickname exchange, direct messaging, group messaging and file sharing. You can check the chat demo.

[drHuangMHT]

I'd recommend HuLa for a polished GUI, but unfortunately it's based on Vue instead of React.

[Lebei2046]

Sure, better for desktop, need to dig into.