rings_core/message/handlers/
mod.rs

1#![warn(missing_docs)]
2//! This module implemented message handler of rings network.
3
4use std::sync::Arc;
5
6use async_recursion::async_recursion;
7use async_trait::async_trait;
8
9use super::effects::lower_dht_action;
10use super::effects::ConnectionFunctor;
11use super::effects::CoreEffect;
12use super::effects::CoreEffectInterpreter;
13use super::MessagePayload;
14use crate::dht::ChordStorageRepair;
15use crate::dht::CorrectChord;
16use crate::dht::Did;
17use crate::dht::PeerRing;
18use crate::dht::PeerRingAction;
19use crate::error::Error;
20use crate::error::Result;
21use crate::swarm::callback::InnerSwarmCallback;
22use crate::swarm::callback::SharedSwarmCallback;
23use crate::swarm::transport::SwarmTransport;
24
25/// Operator and Handler for Connection
26pub mod connection;
27/// Operator and Handler for CustomMessage
28pub mod custom;
29/// Operator and Handler for E2E encrypted messages
30pub mod e2e;
31/// Operator and handler for DHT stabilization
32pub mod stabilization;
33/// Operator and Handler for Storage
34pub mod storage;
35/// Operator and Handler for Subring
36pub mod subring;
37
38/// Shared message-handler handle.
39///
40/// Clone law: cloning duplicates `Arc` handles to the same transport, DHT
41/// state, and callback. It never forks protocol state or transfers ownership.
42#[derive(Clone)]
43pub struct MessageHandler {
44    transport: Arc<SwarmTransport>,
45    dht: Arc<PeerRing>,
46    swarm_callback: SharedSwarmCallback,
47}
48
49/// Generic trait for handle message ,inspired by Actor-Model.
50#[cfg_attr(feature = "wasm", async_trait(?Send))]
51#[cfg_attr(not(feature = "wasm"), async_trait)]
52pub trait HandleMsg<T> {
53    /// Message handler.
54    async fn handle(&self, ctx: &MessagePayload, msg: &T) -> Result<()>;
55}
56
57impl MessageHandler {
58    /// Create a new MessageHandler instance.
59    pub fn new(transport: Arc<SwarmTransport>, swarm_callback: SharedSwarmCallback) -> Self {
60        let dht = transport.dht.clone();
61        Self {
62            transport,
63            dht,
64            swarm_callback,
65        }
66    }
67
68    fn inner_callback(&self) -> InnerSwarmCallback {
69        InnerSwarmCallback::new(self.transport.clone(), self.swarm_callback.clone())
70    }
71
72    pub(crate) async fn run_effects<'payload>(
73        &self,
74        effects: impl IntoIterator<Item = CoreEffect<'payload>>,
75    ) -> Result<()> {
76        CoreEffectInterpreter::new(&self.transport, &self.swarm_callback)
77            .run_all(effects)
78            .await
79    }
80
81    /// Idempotently establish a DHT-driven transport connection.
82    ///
83    /// Self and already-connected peers are no-ops. `AlreadyConnected` is treated
84    /// as success so concurrent DHT actions racing through `MultiActions` do not
85    /// fail the whole handler.
86    pub(crate) async fn connect_dht_peer(&self, peer: Did) -> Result<()> {
87        self.run_effects([ConnectionFunctor::connect_dht_peer(peer).into()])
88            .await
89    }
90
91    pub(crate) async fn join_dht(&self, peer: Did) -> Result<()> {
92        // Default HMCC/Zave join path: maps to the JoinThenSync operation in
93        // the CorrectChord spec (see tests/default/dht_convergence.rs).
94        let conn = self
95            .transport
96            .get_connection(peer)
97            .ok_or(Error::SwarmMissDidInTable(peer))?;
98        let dht_ev = self.dht.join_then_sync(conn).await?;
99        // The local join has completed. Follow-up convergence messages are
100        // best-effort: a peer can churn before these sends complete, and that
101        // must not suppress the application-level Connected event.
102        if let Err(e) = self.handle_dht_events(&dht_ev).await {
103            tracing::warn!("Failed to handle dht events while joining {peer}: {e:?}");
104        }
105        Ok(())
106    }
107
108    pub(crate) async fn leave_dht(&self, peer: Did) -> Result<()> {
109        if self
110            .transport
111            .get_and_check_connection(peer)
112            .await
113            .is_none()
114        {
115            let should_repair = self
116                .dht
117                .peer_may_share_storage_responsibility(peer, self.transport.storage_redundancy())
118                .await?;
119            self.dht.remove(peer)?;
120            if should_repair {
121                let repair = self
122                    .dht
123                    .republish_local_entries(self.transport.storage_redundancy())
124                    .await?;
125                storage::handle_storage_repair_act(self.transport.clone(), repair).await?;
126            }
127        };
128        Ok(())
129    }
130
131    #[cfg_attr(feature = "wasm", async_recursion(?Send))]
132    #[cfg_attr(not(feature = "wasm"), async_recursion)]
133    pub(crate) async fn handle_dht_events(&self, act: &PeerRingAction) -> Result<()> {
134        match act {
135            PeerRingAction::MultiActions(acts) => {
136                let jobs = acts
137                    .iter()
138                    .map(|act| async move { self.handle_dht_events(act).await });
139
140                for res in futures::future::join_all(jobs).await {
141                    if res.is_err() {
142                        tracing::error!("Failed on handle multi actions: {:#?}", res)
143                    }
144                }
145
146                Ok(())
147            }
148            act => {
149                let effects =
150                    lower_dht_action(act, |did| self.transport.get_connection(did).is_some())?;
151                self.run_effects(effects).await
152            }
153        }
154    }
155}
rings_core/message/handlers/mod.rs

rings_core/message/handlers/
mod.rs
