network_libp2p/

network_node.rs

use std::collections::HashMap;

use chat_core::network::{Multiaddr, NetworkEvent, NetworkNode, NetworkResult, RequestId, RoomId};
use chat_core::stats::RawStats;
use libp2p::PeerId;

use crate::common::Libp2pNetwork;

impl NetworkNode for Libp2pNetwork {
    fn local_id(&self) -> chat_core::network::PeerId {
        chat_core::network::PeerId::from(self.local_peer_id.to_string())
    }

    fn dial(&mut self, addr: &Multiaddr) -> NetworkResult<()> {
        let addr: libp2p::Multiaddr = addr
            .as_str()
            .parse()
            .map_err(|e: libp2p::multiaddr::Error| e.to_string())?;
        if let Some(peer_id) = addr.iter().find_map(|p| {
            if let libp2p::multiaddr::Protocol::P2p(id) = p {
                Some(id)
            } else {
                None
            }
        }) {
            if self.pending_dials.contains(&peer_id)
                || self.swarm.behaviour().gossipsub_manager.is_dialing(&peer_id)
            {
                return Err("Already dialing peer".to_string().into());
            }
            self.pending_dials.insert(peer_id);
            if let Err(e) = self.swarm.dial(addr) {
                self.pending_dials.remove(&peer_id);
                return Err(e.to_string().into());
            }
            Ok(())
        } else {
            self.swarm.dial(addr).map_err(|e| e.to_string().into())
        }
    }

    fn join_room(&mut self, room: &RoomId) -> NetworkResult<()> {
        let topic_name = room.as_str();
        let topic = self
            .swarm
            .behaviour_mut()
            .gossipsub_manager
            .subscribe(topic_name)
            .map_err(|e| e.to_string())?;
        let cid = self
            .swarm
            .behaviour()
            .gossipsub_manager
            .topic_to_cid(topic_name)
            .map_err(|e| e.to_string())?;
        let key = libp2p::kad::RecordKey::new(&cid.to_bytes());
        tracing::info!("DHT start_providing for room {}", topic_name);
        self.swarm
            .behaviour_mut()
            .kad
            .start_providing(key.clone())
            .map_err(|e| format!("start_providing error: {:?}", e))?;
        if self.dht_discovery_enabled {
            tracing::info!("DHT get_providers for room {}", topic_name);
            self.swarm.behaviour_mut().kad.get_providers(key);
        }
        self.swarm
            .behaviour_mut()
            .gossipsub
            .subscribe(&topic)
            .map_err(|e| format!("Failed to subscribe to gossipsub topic: {}", e))?;
        Ok(())
    }

    fn publish_message(&mut self, data: Vec<u8>) -> NetworkResult<()> {
        // Publish to all subscribed topics. In practice there is only one active room.
        let topics: Vec<_> = self.swarm.behaviour().gossipsub.topics().cloned().collect();
        let mut any_success = false;
        for topic_hash in &topics {
            match self
                .swarm
                .behaviour_mut()
                .gossipsub
                .publish(topic_hash.clone(), data.clone())
            {
                Ok(_) => any_success = true,
                Err(e) => {
                    tracing::warn!("Gossipsub publish failed for topic {}: {:?}", topic_hash, e);
                }
            }
        }
        if !any_success && !topics.is_empty() {
            return Err("No gossipsub topics available for publish".to_string().into());
        }
        Ok(())
    }

    fn send_message(&mut self, peer_id: &chat_core::network::PeerId, data: Vec<u8>) -> NetworkResult<RequestId> {
        let peer_id: PeerId = match peer_id.as_str().parse() {
            Ok(p) => p,
            Err(e) => return Err(e.to_string().into()),
        };
        let req_id = self.swarm.behaviour_mut().history.send_request(&peer_id, data);
        Ok(RequestId::from(req_id.to_string()))
    }

    fn send_response(&mut self, request_id: &RequestId, data: Vec<u8>) -> NetworkResult<()> {
        let channel = self
            .pending_responses
            .remove(request_id.as_str())
            .ok_or_else(|| "Unknown request id".to_string())?;
        if self
            .swarm
            .behaviour_mut()
            .history
            .send_response(channel, data)
            .is_err()
        {
            return Err("Failed to send response (channel closed)".to_string().into());
        }
        Ok(())
    }

    fn connected_peers(&self) -> Vec<chat_core::network::PeerId> {
        self.swarm
            .connected_peers()
            .filter(|p| *p != &self.local_peer_id)
            .map(|p| chat_core::network::PeerId::from(p.to_string()))
            .collect()
    }

    fn is_dialing(&self, peer_id: &chat_core::network::PeerId) -> bool {
        match peer_id.as_str().parse::<PeerId>() {
            Ok(p) => {
                self.pending_dials.contains(&p)
                    || self.swarm.behaviour().gossipsub_manager.is_dialing(&p)
            }
            Err(_) => false,
        }
    }

    fn raw_stats(&self) -> RawStats {
        let num_peers = self.swarm.network_info().num_peers() as u32;
        let relay_count = self
            .swarm
            .external_addresses()
            .filter(|a| a.to_string().contains("p2p-circuit"))
            .count() as u32;

        let addrs = self.dial_addrs();
        let circuit_address = addrs.iter().find(|a| a.as_str().contains("p2p-circuit")).map(|a| a.as_str().to_string());
        let web_rtc_address = addrs.iter().find(|a| a.as_str().contains("webrtc")).map(|a| a.as_str().to_string());

        let listeners: Vec<String> = self.swarm.listeners().map(|a| a.to_string()).collect();
        let external_addresses: Vec<String> =
            self.swarm.external_addresses().map(|a| a.to_string()).collect();

        let connected_peers: Vec<String> = self
            .swarm
            .connected_peers()
            .filter(|p| *p != &self.local_peer_id)
            .map(|p| p.to_string())
            .collect();

        let mut peer_conn_addrs = HashMap::new();
        let mut peer_advertised_addrs = HashMap::new();
        for peer_id in &connected_peers {
            let pid: PeerId = match peer_id.parse() {
                Ok(p) => p,
                Err(_) => continue,
            };
            let conn_addrs = self.peer_conn_addrs.get(&pid).cloned().unwrap_or_default();
            peer_conn_addrs.insert(peer_id.clone(), conn_addrs);
            let advertised: Vec<String> = self.peer_addrs.get(&pid).cloned().unwrap_or_default().iter().map(|a| a.to_string()).collect();
            peer_advertised_addrs.insert(peer_id.clone(), advertised);
        }

        RawStats {
            peer_id: self.local_peer_id.to_string(),
            num_peers,
            relay_count,
            listeners,
            external_addresses,
            connected_peers,
            peer_conn_addrs,
            peer_advertised_addrs,
            circuit_address,
            web_rtc_address,
        }
    }

    fn bootstrap(&mut self) -> NetworkResult<()> {
        self.swarm
            .behaviour_mut()
            .kad
            .bootstrap()
            .map(|_| ())
            .map_err(|e| e.to_string().into())
    }

    fn dial_addrs(&self) -> Vec<Multiaddr> {
        self.platform_dial_addrs()
    }

    async fn next_event(&mut self) -> Option<NetworkEvent> {
        self.platform_next_event().await
    }
}