| 283 | /// Starts the libp2p service networking stack. This Future resolves when shutdown occurs. |
| 284 | #[instrument(skip_all)] |
| 285 | pub async fn run(&mut self) -> Result<()> { |
| 286 | info!("P2P listen addrs: {:?}", self.listen_addrs()); |
| 287 | info!("Local Peer ID: {}", self.local_peer_id()); |
| 288 | |
| 289 | let mut nice_interval = self.use_dht.then(|| tokio::time::interval(NICE_INTERVAL)); |
| 290 | // Initialize bootstrap_interval to not start immediately but at now + interval. |
| 291 | // This is because we know that initially there are no nodes with whom to bootstrap. |
| 292 | // This interval can be reset if we find a kademlia node before the first tick. |
| 293 | let mut bootstrap_interval = |
| 294 | tokio::time::interval_at(Instant::now() + BOOTSTRAP_INTERVAL, BOOTSTRAP_INTERVAL); |
| 295 | let mut expiry_interval = tokio::time::interval(EXPIRY_INTERVAL); |
| 296 | |
| 297 | #[derive(Debug)] |
| 298 | enum KadBootstrapState { |
| 299 | // Kademlia is idle as it does not have any peers to communicate with. |
| 300 | Idle, |
| 301 | // Kademlia has begun the process of bootstrapping with at least one peer. |
| 302 | Bootstrapping, |
| 303 | // Kademlia has finished the bootstrap process. |
| 304 | Bootstrapped, |
| 305 | } |
| 306 | |
| 307 | let mut kad_state = KadBootstrapState::Idle; |
| 308 | loop { |
| 309 | self.metrics.record(&LoopEvent); |
| 310 | tokio::select! { |
| 311 | swarm_event = self.swarm.next() => { |
| 312 | let swarm_event = swarm_event.expect("the swarm will never die"); |
| 313 | match self.handle_swarm_event(swarm_event).await { |
| 314 | Ok(Some(SwarmEventResult::KademliaBoostrapSuccess)) => { |
| 315 | kad_state = KadBootstrapState::Bootstrapped; |
| 316 | } |
| 317 | Ok(Some(SwarmEventResult::KademliaAddressAdded)) => { |
| 318 | if matches!(kad_state, KadBootstrapState::Idle) { |
| 319 | kad_state = KadBootstrapState::Bootstrapping; |
| 320 | bootstrap_interval.reset_immediately(); |
| 321 | } |
| 322 | } |
| 323 | Ok(None) => {}, |
| 324 | Err(err) => error!("swarm error: {:?}",err), |
| 325 | }; |
| 326 | |
| 327 | if let Some(kad) = self.swarm.behaviour_mut().kad.as_mut() { |
| 328 | self.providers.poll(kad); |
| 329 | } |
| 330 | } |
| 331 | rpc_message = self.net_receiver_in.recv() => { |
| 332 | match rpc_message { |
| 333 | Some(rpc_message) => { |
| 334 | match self.handle_rpc_message(rpc_message).await { |
| 335 | Ok(true) => { |
| 336 | // shutdown |
| 337 | return Ok(()); |
| 338 | } |
| 339 | Ok(false) => { |
| 340 | continue; |
| 341 | } |
| 342 | Err(err) => { |