rings_core/swarm/
mod.rs

1#![deny(missing_docs)]
2
3//! This mod is the main entrance of swarm.
4
5mod builder;
6/// Callback interface for swarm
7pub mod callback;
8pub(crate) mod transport;
9
10use std::num::NonZeroUsize;
11use std::sync::Arc;
12use std::sync::RwLock;
13
14pub use builder::SwarmBuilder;
15
16use self::callback::InnerSwarmCallback;
17use crate::dht::Did;
18use crate::dht::PeerRing;
19use crate::dht::Stabilizer;
20use crate::ecc::PublicKey;
21use crate::ecc::VerificationPublicKey;
22use crate::error::Error;
23use crate::error::Result;
24use crate::inspect::ConnectionInspect;
25use crate::inspect::SwarmInspect;
26use crate::measure::PeerMeasurement;
27use crate::measure::PeerMeasurementPage;
28use crate::message::DhtProtocolMode;
29use crate::message::Message;
30use crate::message::MessagePayload;
31use crate::message::MessageVerificationExt;
32use crate::message::PayloadSender;
33use crate::swarm::callback::SharedSwarmCallback;
34use crate::swarm::transport::SwarmTransport;
35
36/// The transport and dht management.
37pub struct Swarm {
38    /// Reference of DHT.
39    pub(crate) dht: Arc<PeerRing>,
40    /// Swarm transport.
41    pub(crate) transport: Arc<SwarmTransport>,
42    callback: RwLock<SharedSwarmCallback>,
43}
44
45impl Swarm {
46    /// Get did of self.
47    pub fn did(&self) -> Did {
48        self.dht.did
49    }
50
51    /// Get the local account public key used for E2E public-key negotiation.
52    pub fn account_pubkey(&self) -> Result<PublicKey<33>> {
53        self.transport.session_sk().session().account_pubkey()
54    }
55
56    /// Get the typed account verification public key.
57    pub fn account_verification_pubkey(&self) -> Result<VerificationPublicKey> {
58        self.transport
59            .session_sk()
60            .session()
61            .account_verification_pubkey()
62    }
63
64    /// Get this swarm's network id.
65    pub fn network_id(&self) -> u32 {
66        self.transport.network_id
67    }
68
69    /// Get the storage redundancy for this swarm's DHT protocol mode.
70    pub fn storage_redundancy(&self) -> u16 {
71        self.transport.storage_redundancy()
72    }
73
74    /// Get the storage virtual-node positions for this swarm's DHT protocol mode.
75    pub fn dht_virtual_nodes(&self) -> u16 {
76        self.transport.dht_virtual_nodes()
77    }
78
79    /// Get this swarm's full DHT protocol mode descriptor.
80    pub fn dht_protocol_mode(&self) -> DhtProtocolMode {
81        self.transport.dht_protocol_mode()
82    }
83
84    /// Get DHT(Distributed Hash Table) of self.
85    pub fn dht(&self) -> Arc<PeerRing> {
86        self.dht.clone()
87    }
88
89    fn callback(&self) -> Result<SharedSwarmCallback> {
90        Ok(self
91            .callback
92            .read()
93            .map_err(|_| Error::CallbackSyncLockError)?
94            .clone())
95    }
96
97    fn inner_callback(&self) -> Result<InnerSwarmCallback> {
98        Ok(InnerSwarmCallback::new(
99            self.transport.clone(),
100            self.callback()?,
101        ))
102    }
103
104    /// Set callback for swarm.
105    pub fn set_callback(&self, callback: SharedSwarmCallback) -> Result<()> {
106        let mut inner = self
107            .callback
108            .write()
109            .map_err(|_| Error::CallbackSyncLockError)?;
110
111        *inner = callback;
112
113        Ok(())
114    }
115
116    /// Create [Stabilizer] for swarm.
117    pub fn stabilizer(&self) -> Stabilizer {
118        Stabilizer::new(self.transport.clone())
119    }
120
121    /// Disconnect a connection. There are three steps:
122    /// 1) remove from DHT;
123    /// 2) remove from Transport;
124    /// 3) close the connection;
125    pub async fn disconnect(&self, peer: Did) -> Result<()> {
126        self.transport.disconnect(peer).await
127    }
128
129    /// Start a non-routable handshake with a peer.
130    ///
131    /// The peer becomes visible through the connection inspection APIs only
132    /// after its data channel opens and the swarm admits it to the DHT.
133    pub async fn connect(&self, peer: Did) -> Result<()> {
134        if peer == self.did() {
135            return Err(Error::ShouldNotConnectSelf);
136        }
137        self.transport.connect(peer, self.inner_callback()?).await
138    }
139
140    /// Send [Message] to peer.
141    pub async fn send_message(&self, msg: Message, destination: Did) -> Result<uuid::Uuid> {
142        self.transport.send_message(msg, destination).await
143    }
144
145    /// Send a message directly to an already connected peer, without a Chord lookup.
146    ///
147    /// This preserves an application protocol's explicit next-hop selection. Callers must
148    /// ensure the destination has an active direct transport connection.
149    pub async fn send_direct_message(&self, msg: Message, destination: Did) -> Result<uuid::Uuid> {
150        self.transport.send_direct_message(msg, destination).await
151    }
152
153    /// List active, routable peers and their connection status.
154    pub fn peers(&self) -> Vec<ConnectionInspect> {
155        self.transport
156            .get_connections()
157            .iter()
158            .map(|(did, c)| ConnectionInspect {
159                did: did.to_string(),
160                state: format!("{:?}", c.webrtc_connection_state()),
161            })
162            .collect()
163    }
164
165    /// List DIDs with active, routable transport connections.
166    pub fn peer_dids(&self) -> Vec<Did> {
167        self.transport.get_connection_ids()
168    }
169
170    /// List DIDs whose direct WebRTC transport connection is active.
171    pub fn connected_peer_dids(&self) -> Vec<Did> {
172        self.transport.get_connection_ids()
173    }
174
175    /// Return local measurement counters for `peer`, if observed.
176    pub async fn peer_measurement(&self, peer: Did) -> Option<PeerMeasurement> {
177        self.transport.peer_measurement(peer).await
178    }
179
180    /// Return every retained local peer measurement.
181    pub async fn peer_measurements(&self) -> Vec<PeerMeasurement> {
182        self.transport.peer_measurements().await
183    }
184
185    /// Return one bounded page of retained local peer measurements.
186    pub async fn peer_measurements_page(
187        &self,
188        after: Option<Did>,
189        limit: NonZeroUsize,
190    ) -> PeerMeasurementPage {
191        self.transport.peer_measurements_page(after, limit).await
192    }
193
194    /// Check the status of swarm
195    pub async fn inspect(&self) -> SwarmInspect {
196        SwarmInspect::inspect(self).await
197    }
198}
199
200impl Swarm {
201    /// Create new connection and its answer. This function will wrap the offer inside a payload
202    /// with verification.
203    pub async fn create_offer(&self, peer: Did) -> Result<MessagePayload> {
204        let offer_msg = self
205            .transport
206            .prepare_connection_offer(peer, self.inner_callback()?)
207            .await?;
208
209        // This payload has fake next_hop.
210        // The invoker should fix it before sending.
211        let payload = MessagePayload::new_send(
212            Message::ConnectNodeSend(offer_msg),
213            self.transport.session_sk(),
214            self.did(),
215            peer,
216        )?;
217
218        Ok(payload)
219    }
220
221    /// Answer the offer of remote connection. This function will verify the answer payload and
222    /// will wrap the answer inside a payload with verification.
223    pub async fn answer_offer(&self, offer_payload: MessagePayload) -> Result<MessagePayload> {
224        if !offer_payload.verify() {
225            return Err(Error::VerifySignatureFailed);
226        }
227
228        let Message::ConnectNodeSend(msg) = offer_payload.transaction.data()? else {
229            return Err(Error::InvalidMessage(
230                "Should be ConnectNodeSend".to_string(),
231            ));
232        };
233
234        let peer = offer_payload.transaction.signer();
235        let answer_msg = self
236            .transport
237            .answer_remote_connection(peer, self.inner_callback()?, &msg)
238            .await?;
239
240        // This payload has fake next_hop.
241        // The invoker should fix it before sending.
242        let answer_payload = MessagePayload::new_send(
243            Message::ConnectNodeReport(answer_msg),
244            self.transport.session_sk(),
245            self.did(),
246            self.did(),
247        )?;
248
249        Ok(answer_payload)
250    }
251
252    /// Accept the answer of remote connection. This function will verify the answer payload and
253    /// will return its did with the connection.
254    pub async fn accept_answer(&self, answer_payload: MessagePayload) -> Result<()> {
255        if !answer_payload.verify() {
256            return Err(Error::VerifySignatureFailed);
257        }
258
259        let Message::ConnectNodeReport(ref msg) = answer_payload.transaction.data()? else {
260            return Err(Error::InvalidMessage(
261                "Should be ConnectNodeReport".to_string(),
262            ));
263        };
264
265        let peer = answer_payload.transaction.signer();
266        self.transport.accept_remote_connection(peer, msg).await
267    }
268}
rings_core/swarm/mod.rs

rings_core/swarm/
mod.rs
