hopr_transport_p2p/
lib.rs

1//! # P2P
2//!
3//! The underlying technology for managing the peer-to-peer networking used by this package is the [`rust-libp2p`](https://github.com/libp2p/rust-libp2p) library ([documentation](https://docs.libp2p.io/)).
4//!
5//! ## Modularity
6//!
7//! `rust-libp2p` is highly modular allowing for reimplmenting expected behavior using custom implementations for API
8//! traits.
9//!
10//! This way it is possible to experiment with and combine different components of the library in order to construct a
11//! specific targeted use case.
12//!
13//! ## `rust-libp2p` connectivity
14//!
15//! As per the [official documentation](https://connectivity.libp2p.io/), the connectivity types in the library are divided into the `standalone` (implementation of network over host) and `browser` (implementation of network over browser).
16//!
17//! Nodes that are not located behind a blocking firewall or NAT are designated as **public nodes** and can utilize the
18//! `TCP` or `QUIC` connectivity, with the recommendation to use QUIC if possible.
19//!
20//! Browser based solutions are almost always located behind a private network or a blocking firewall and to open a
21//! connection towards the standalone nodes these utilize either the `WebSocket` approach (by hijacking the `TCP`
22//! connection) or the (not yet fully speced up) `WebTransport` (by hijacking the `QUIC` connection).
23
24/// Constants exported by the crate.
25pub mod constants;
26
27/// Errors generated by the crate.
28pub mod errors;
29
30/// Per-peer connection liveness tracking for libp2p streams.
31pub(crate) mod liveness;
32
33/// In-memory peer store for network multiaddresses.
34pub mod peer_store;
35
36/// Raw swarm definition for the HOPR network.
37pub mod swarm;
38
39/// P2P behavior definitions for the transport level interactions not related to the HOPR protocol
40mod behavior;
41
42use std::{collections::HashSet, sync::Arc};
43
44use dashmap::DashSet;
45use futures::{AsyncRead, AsyncWrite, StreamExt};
46pub use hopr_api::network::Health;
47use hopr_api::network::{NetworkView, traits::NetworkStreamControl};
48use libp2p::{Multiaddr, PeerId};
49
50use crate::liveness::{LivenessRegistry, LivenessStream};
51
52mod utils;
53
54pub use crate::{
55    behavior::{HoprNetworkBehavior, HoprNetworkBehaviorEvent},
56    swarm::HoprLibp2pNetworkBuilder,
57};
58
59/// Events required for bootstrapping the p2p network.
60#[derive(Debug, Clone)]
61pub enum PeerDiscovery {
62    Announce(PeerId, Vec<Multiaddr>),
63}
64
65#[derive(Clone)]
66pub struct HoprNetwork {
67    tracker: Arc<DashSet<PeerId>>,
68    store: Arc<crate::peer_store::NetworkPeerStore>,
69    control: libp2p_stream::Control,
70    protocol: libp2p::StreamProtocol,
71    event_rx: async_broadcast::InactiveReceiver<hopr_api::network::NetworkEvent>,
72    /// Shared liveness registry. Entries are cleared by the swarm event loop
73    /// on `ConnectionClosed` / `OutgoingConnectionError` so that wrapped
74    /// substreams for that peer self-error on their next poll.
75    liveness: LivenessRegistry,
76}
77
78impl std::fmt::Debug for HoprNetwork {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct("HoprNetwork")
81            .field("tracker", &self.tracker)
82            .field("store", &self.store)
83            .field("protocol", &self.protocol)
84            .finish_non_exhaustive()
85    }
86}
87
88impl NetworkView for HoprNetwork {
89    fn listening_as(&self) -> HashSet<Multiaddr> {
90        self.store.get(self.store.me()).unwrap_or_else(|| {
91            tracing::error!("failed to get own peer info from the peer store");
92            std::collections::HashSet::new()
93        })
94    }
95
96    #[inline]
97    fn discovered_peers(&self) -> HashSet<PeerId> {
98        self.store.iter_keys().collect()
99    }
100
101    #[inline]
102    fn connected_peers(&self) -> HashSet<PeerId> {
103        self.tracker.iter().map(|r| *r).collect()
104    }
105
106    fn is_connected(&self, peer: &PeerId) -> bool {
107        self.tracker.contains(peer)
108    }
109
110    #[inline]
111    fn multiaddress_of(&self, peer: &PeerId) -> Option<HashSet<Multiaddr>> {
112        self.store.get(peer)
113    }
114
115    fn health(&self) -> Health {
116        match self.tracker.len() {
117            0 => Health::Red,
118            1 => Health::Orange,
119            2..4 => Health::Yellow,
120            _ => Health::Green,
121        }
122    }
123
124    fn subscribe_network_events(
125        &self,
126    ) -> impl futures::Stream<Item = hopr_api::network::NetworkEvent> + Send + 'static {
127        self.event_rx.clone().activate()
128    }
129}
130
131#[async_trait::async_trait]
132impl NetworkStreamControl for HoprNetwork {
133    fn accept(
134        mut self,
135    ) -> Result<impl futures::Stream<Item = (PeerId, impl AsyncRead + AsyncWrite + Send)> + Send, impl std::error::Error>
136    {
137        let liveness = self.liveness.clone();
138        self.control.accept(self.protocol).map(|stream| {
139            stream.map(move |(peer, inner)| {
140                let flag = liveness.get_or_create_connected(&peer);
141                (peer, LivenessStream::new(inner, flag))
142            })
143        })
144    }
145
146    async fn open(mut self, peer: PeerId) -> Result<impl AsyncRead + AsyncWrite + Send, impl std::error::Error> {
147        let flag = self.liveness.get_or_create_connected(&peer);
148        self.control
149            .open_stream(peer, self.protocol)
150            .await
151            .map(|inner| LivenessStream::new(inner, flag))
152    }
153}
hopr_transport_p2p/lib.rs

hopr_transport_p2p/
lib.rs
