rings_node/processor/
builder.rs

1use super::config::validate_onion_role_config;
2use super::*;
3
4/// ProcessorBuilder is used to initialize a [Processor] instance.
5pub struct ProcessorBuilder {
6    pub(in crate::processor) network_id: u32,
7    pub(in crate::processor) ice_servers: String,
8    pub(in crate::processor) external_address: Option<String>,
9    pub(in crate::processor) webrtc_udp_port_range: Option<WebrtcUdpPortRange>,
10    pub(in crate::processor) session_sk: SessionSk,
11    pub(in crate::processor) storage: Option<EntryStorage>,
12    pub(in crate::processor) measure: Option<Arc<PeriodicMeasure>>,
13    pub(in crate::processor) stabilize_interval: Duration,
14    pub(in crate::processor) online_node_heartbeat_interval: Duration,
15    pub(in crate::processor) online_node_ttl: Duration,
16    pub(in crate::processor) online_node_type: OnlineNodeType,
17    pub(in crate::processor) advertise_presence: bool,
18    pub(in crate::processor) dht_virtual_nodes: u16,
19    pub(in crate::processor) advertise_onion_relay: bool,
20    pub(in crate::processor) advertise_onion_exit: bool,
21    pub(in crate::processor) onion_exit_heartbeat_interval: Duration,
22    pub(in crate::processor) onion_exit_ttl: Duration,
23    pub(in crate::processor) onion_exit_services: Vec<OnionExitService>,
24    pub(in crate::processor) onion_exit_policy: OnionExitPolicy,
25    pub(in crate::processor) registration_tasks: Vec<Arc<dyn RegistrationTask>>,
26    pub(in crate::processor) dht_finger_table_size: usize,
27    pub(in crate::processor) reassembly_limits: ReassemblyLimits,
28}
29
30impl ProcessorBuilder {
31    /// initialize a [ProcessorBuilder] with a serialized [ProcessorConfig].
32    pub fn from_serialized(config: &str) -> Result<Self> {
33        let config =
34            serde_yaml::from_str::<ProcessorConfig>(config).map_err(Error::SerdeYamlError)?;
35        Self::from_config(&config)
36    }
37
38    /// initialize a [ProcessorBuilder] with a [ProcessorConfig].
39    pub fn from_config(config: &ProcessorConfig) -> Result<Self> {
40        validate_online_node_registration_timing(
41            config.advertise_presence,
42            config.online_node_heartbeat_interval,
43            config.online_node_ttl,
44        )?;
45        validate_onion_exit_registration_timing(
46            config.advertise_onion_exit,
47            config.onion_exit_heartbeat_interval,
48            config.onion_exit_ttl,
49        )?;
50        validate_onion_role_config(
51            config.advertise_presence,
52            config.advertise_onion_relay,
53            config.advertise_onion_exit,
54            &config.onion_exit_services,
55            &config.onion_exit_policy,
56        )?;
57        Ok(Self {
58            network_id: config.network_id,
59            ice_servers: config.ice_servers.clone(),
60            external_address: config.external_address.clone(),
61            webrtc_udp_port_range: config.webrtc_udp_port_range()?,
62            session_sk: config.session_sk.clone(),
63            storage: None,
64            measure: None,
65            stabilize_interval: config.stabilize_interval,
66            online_node_heartbeat_interval: config.online_node_heartbeat_interval,
67            online_node_ttl: config.online_node_ttl,
68            online_node_type: config.online_node_type.clone(),
69            advertise_presence: config.advertise_presence,
70            dht_virtual_nodes: config.dht_virtual_nodes,
71            advertise_onion_relay: config.advertise_onion_relay,
72            advertise_onion_exit: config.advertise_onion_exit,
73            onion_exit_heartbeat_interval: config.onion_exit_heartbeat_interval,
74            onion_exit_ttl: config.onion_exit_ttl,
75            onion_exit_services: config.onion_exit_services.clone(),
76            onion_exit_policy: config.onion_exit_policy.clone(),
77            registration_tasks: Vec::new(),
78            dht_finger_table_size: DEFAULT_FINGER_TABLE_SIZE,
79            reassembly_limits: ReassemblyLimits::production(),
80        })
81    }
82
83    /// Set the storage for the processor.
84    pub fn storage(mut self, storage: EntryStorage) -> Self {
85        self.storage = Some(storage);
86        self
87    }
88
89    /// Set the measure for the processor.
90    pub fn measure(mut self, implement: PeriodicMeasure) -> Self {
91        self.measure = Some(Arc::new(implement));
92        self
93    }
94
95    /// Set the number of DHT finger-table slots for the processor's swarm.
96    pub fn dht_finger_table_size(mut self, size: usize) -> Self {
97        self.dht_finger_table_size = size;
98        self
99    }
100
101    /// Set storage-only virtual positions derived per physical peer.
102    ///
103    /// Serialized configs reject values above
104    /// [`rings_core::dht::MAX_STORAGE_VIRTUAL_POSITIONS_PER_OWNER`]. This
105    /// builder setter is infallible for direct programmatic use; the core swarm
106    /// builder normalizes the value once before storage ownership and protocol
107    /// advertisement are created.
108    pub fn dht_virtual_nodes(mut self, positions_per_peer: u16) -> Self {
109        self.dht_virtual_nodes = positions_per_peer;
110        self
111    }
112
113    /// Set inbound chunk reassembly limits for the processor's swarm.
114    pub fn reassembly_limits(mut self, limits: ReassemblyLimits) -> Self {
115        self.reassembly_limits = limits;
116        self
117    }
118
119    /// Set the runtime family advertised in the online-node registry.
120    pub fn online_node_type(mut self, node_type: OnlineNodeType) -> Self {
121        self.online_node_type = node_type;
122        self
123    }
124
125    /// Set whether listen() advertises this node's presence.
126    pub fn advertise_presence(mut self, advertise: bool) -> Self {
127        self.advertise_presence = advertise;
128        self
129    }
130
131    /// Set whether listen() advertises this node as an onion relay.
132    pub fn advertise_onion_relay(mut self, advertise: bool) -> Self {
133        self.advertise_onion_relay = advertise;
134        self
135    }
136
137    /// Set whether listen() publishes this node as an onion exit.
138    pub fn advertise_onion_exit(mut self, advertise: bool) -> Self {
139        self.advertise_onion_exit = advertise;
140        self
141    }
142
143    /// Add a custom periodic registration task.
144    pub fn registration_task<T>(mut self, task: T) -> Self
145    where T: RegistrationTask + 'static {
146        self.registration_tasks.push(Arc::new(task));
147        self
148    }
149
150    /// Add an already shared custom periodic registration task.
151    pub fn shared_registration_task(mut self, task: Arc<dyn RegistrationTask>) -> Self {
152        self.registration_tasks.push(task);
153        self
154    }
155
156    /// Build the [Processor].
157    pub fn build(self) -> Result<Processor> {
158        self.session_sk
159            .session()
160            .verify_self()
161            .map_err(|e| Error::VerifyError(e.to_string()))?;
162
163        let storage = self.storage.unwrap_or_else(|| Box::new(MemStorage::new()));
164        let endpoint_hint = self.external_address.clone();
165        let mut online_node_capabilities = Vec::new();
166        if self.advertise_onion_relay {
167            online_node_capabilities.push(ONION_RELAY_CAPABILITY.to_string());
168        }
169        let session_sk = self.session_sk.clone();
170        let online_node_registration = OnlineNodeRegistration::new(
171            self.online_node_heartbeat_interval,
172            self.online_node_ttl,
173            self.online_node_type.clone(),
174            endpoint_hint,
175            online_node_capabilities,
176        );
177        let mut registration_tasks = self.registration_tasks;
178        if self.advertise_presence {
179            online_node_registration.validate_enabled_schedule()?;
180            registration_tasks.push(Arc::new(online_node_registration.clone()));
181        }
182        if self.advertise_onion_exit {
183            let onion_exit_registration = OnionExitRegistration::new(
184                self.onion_exit_heartbeat_interval,
185                self.onion_exit_ttl,
186                self.online_node_type,
187                self.onion_exit_services,
188                self.onion_exit_policy,
189            );
190            onion_exit_registration.validate_enabled_schedule()?;
191            registration_tasks.push(Arc::new(onion_exit_registration));
192        }
193
194        let mut swarm_builder =
195            SwarmBuilder::new(self.network_id, &self.ice_servers, storage, self.session_sk);
196        swarm_builder = swarm_builder.dht_storage_redundancy(DATA_REDUNDANT);
197        swarm_builder = swarm_builder.dht_finger_table_size(self.dht_finger_table_size);
198        swarm_builder = swarm_builder.dht_virtual_nodes(self.dht_virtual_nodes);
199        swarm_builder = swarm_builder.reassembly_limits(self.reassembly_limits);
200
201        if let Some(external_address) = self.external_address {
202            swarm_builder = swarm_builder.external_address(external_address);
203        }
204        if let Some(range) = self.webrtc_udp_port_range {
205            swarm_builder = swarm_builder.webrtc_udp_port_range(range);
206        }
207
208        let measure = self.measure;
209        if let Some(measure) = &measure {
210            let implementation: MeasureImpl = measure.clone();
211            swarm_builder = swarm_builder.measure(implementation);
212        }
213        let swarm = Arc::new(swarm_builder.build());
214
215        Ok(Processor {
216            swarm,
217            session_sk,
218            stabilize_interval: self.stabilize_interval,
219            online_node_registration,
220            measure,
221            #[cfg(all(feature = "browser", target_family = "wasm"))]
222            advertise_onion_relay: self.advertise_onion_relay,
223            registration_tasks,
224        })
225    }
226}
rings_node/processor/builder.rs

rings_node/processor/
builder.rs
