calimero_network/
lib.rs

1#![allow(
2    clippy::allow_attributes,
3    reason = "Needed for lints that don't follow expect"
4)]
5#![expect(
6    clippy::multiple_inherent_impl,
7    reason = "Currently necessary due to code structure"
8)]
9use std::collections::hash_map::HashMap;
10use std::sync::Arc;
11
12use actix::{Actor, AsyncContext, Context};
13use calimero_network_primitives::config::NetworkConfig;
14use calimero_network_primitives::messages::{NetworkEvent, NetworkEventDispatcher};
15use calimero_network_primitives::stream::{CALIMERO_BLOB_PROTOCOL, CALIMERO_STREAM_PROTOCOL};
16use calimero_utils_actix::actor;
17use eyre::Result as EyreResult;
18use futures_util::StreamExt;
19use libp2p::kad::QueryId;
20use libp2p::swarm::Swarm;
21use libp2p::PeerId;
22use libp2p_metrics::Metrics;
23use prometheus_client::registry::Registry;
24use tokio::sync::oneshot;
25use tokio::time::interval;
26use tokio_stream::wrappers::IntervalStream;
27use tracing::error;
28
29use crate::handlers::stream::incoming::FromIncoming;
30
31pub use calimero_network_primitives::autonat_v2 as autonat;
32pub mod behaviour;
33mod discovery;
34mod handlers;
35
36use behaviour::Behaviour;
37use discovery::Discovery;
38use handlers::stream::rendezvous::RendezvousTick;
39use handlers::stream::swarm::FromSwarm;
40
41#[expect(
42    missing_debug_implementations,
43    reason = "Swarm doesn't implement Debug"
44)]
45pub struct NetworkManager {
46    swarm: Box<Swarm<Behaviour>>,
47    event_dispatcher: Arc<dyn NetworkEventDispatcher>,
48    discovery: Discovery,
49    pending_dial: HashMap<PeerId, oneshot::Sender<EyreResult<()>>>,
50    pending_bootstrap: HashMap<QueryId, oneshot::Sender<EyreResult<()>>>,
51    pending_blob_queries: HashMap<QueryId, oneshot::Sender<eyre::Result<Vec<PeerId>>>>,
52    metrics: Metrics,
53}
54
55impl NetworkManager {
56    /// Create a new NetworkManager with an event dispatcher.
57    ///
58    /// The dispatcher receives all network events (gossipsub messages, streams, etc.)
59    /// and must implement `NetworkEventDispatcher` for reliable delivery.
60    pub async fn new(
61        config: &NetworkConfig,
62        event_dispatcher: Arc<dyn NetworkEventDispatcher>,
63        prom_registry: &mut Registry,
64    ) -> eyre::Result<Self> {
65        let swarm = Behaviour::build_swarm(config)?;
66
67        let discovery = Discovery::new(
68            &config.discovery.rendezvous,
69            &config.discovery.relay,
70            &config.discovery.autonat,
71            if config.discovery.advertise_address {
72                &config.swarm.listen
73            } else {
74                &[]
75            },
76        )
77        .await?;
78
79        let this = Self {
80            swarm: Box::new(swarm),
81            event_dispatcher,
82            discovery,
83            pending_dial: HashMap::default(),
84            pending_bootstrap: HashMap::default(),
85            pending_blob_queries: HashMap::new(),
86            metrics: Metrics::new(prom_registry),
87        };
88
89        Ok(this)
90    }
91}
92
93impl Actor for NetworkManager {
94    type Context = Context<Self>;
95
96    actor!(NetworkManager => {
97        .swarm as FromSwarm
98    });
99
100    fn started(&mut self, ctx: &mut Context<Self>) {
101        let mut control = self.swarm.behaviour().stream.new_control();
102
103        match control.accept(CALIMERO_STREAM_PROTOCOL) {
104            Ok(incoming_streams) => {
105                let _inoming_streams_handle =
106                    ctx.add_stream(incoming_streams.map(|(peer_id, stream)| {
107                        FromIncoming::from_stream(peer_id, stream, CALIMERO_STREAM_PROTOCOL)
108                    }));
109            }
110            Err(err) => {
111                error!("Failed to setup control for stream protocol: {:?}", err);
112            }
113        }
114
115        match control.accept(CALIMERO_BLOB_PROTOCOL) {
116            Ok(incoming_blob_streams) => {
117                let _incoming_blob_streams_handle =
118                    ctx.add_stream(incoming_blob_streams.map(|(peer_id, stream)| {
119                        FromIncoming::from_stream(peer_id, stream, CALIMERO_BLOB_PROTOCOL)
120                    }));
121            }
122            Err(err) => {
123                error!("Failed to setup control for blob protocol: {:?}", err);
124            }
125        }
126
127        let _ping_handle = ctx.add_stream(
128            IntervalStream::new(interval(
129                self.discovery.rendezvous_config.discovery_interval,
130            ))
131            .map(RendezvousTick::from),
132        );
133    }
134}
calimero_network/lib.rs

calimero_network/
lib.rs
