async fn initialize_cells(&self, cells: Vec<(Cell, InitialQueueTriggers)>) {
        let (cells, triggers): (Vec<_>, Vec<_>) = cells.into_iter().unzip();
        let networks: Vec<_> = cells
            .iter()
            .map(|cell| cell.aingle_p2p_cell().clone())
            .collect();
        // Add the cells to the conductor map.
        // This write lock can't be held while join is awaited as join calls the conductor.
        // Cells need to be in the map before join is called so it can route the call.
        self.conductor.write().await.add_cells(cells);
        // Join the network but ignore errors because the
        // space retries joining all cells every 5 minutes.
        futures::stream::iter(networks)
            .for_each_concurrent(100, |mut network| async move {
                match tokio::time::timeout(JOIN_NETWORK_TIMEOUT, network.join()).await {
                    Ok(Err(e)) => {
                        tracing::info!(failed_to_join_network = ?e);
                    }
                    Err(_) => {
                        tracing::info!("Timed out trying to join the network");
                    }
                    Ok(Ok(_)) => (),
                }
            })
            .await;

        // Now we can trigger the workflows.
        for trigger in triggers {
            trigger.initialize_workflows();
        }
    }
/// Add cells to the map then join the network then initialize workflows.
Add cells to the map then join the network then initialize workflows.
Add cells to the map then join the network then initialize workflows .
async fn initialize_cells(&self, cells: Vec<(Cell, InitialQueueTriggers)>) {
        let (cells, triggers): (Vec<_>, Vec<_>) = cells.into_iter().unzip();
        let networks: Vec<_> = cells
            .iter()
            .map(|cell| cell.aingle_p2p_cell().clone())
            .collect();
        // Add the cells to the conductor map.
        // This write lock can't be held while join is awaited as join calls the conductor.
        // Cells need to be in the map before join is called so it can route the call.
        self.conductor.write().await.add_cells(cells);
        // Join the network but ignore errors because the
        // space retries joining all cells every 5 minutes.
        futures::stream::iter(networks)
            .for_each_concurrent(100, |mut network| async move {
                match tokio::time::timeout(JOIN_NETWORK_TIMEOUT, network.join()).await {
                    Ok(Err(e)) => {
                        tracing::info!(failed_to_join_network = ?e);
                    }
                    Err(_) => {
                        tracing::info!("Timed out trying to join the network");
                    }
                    Ok(Ok(_)) => (),
                }
            })
            .await;
        // Now we can trigger the workflows.
        for trigger in triggers {
            trigger.initialize_workflows();
        }
    }
async fn initialize_cells ( & self , cells : Vec < ( Cell , InitialQueueTriggers ) > ) { let ( cells , triggers ) : ( Vec < _ > , Vec < _ > ) = cells . into_iter ( ) . unzip ( ) ; let networks : Vec < _ > = cells . iter ( ) . map ( | cell | cell . aingle_p2p_cell ( ) . clone ( ) ) . collect ( ) ; // Add the cells to the conductor map. // This write lock can't be held while join is awaited as join calls the conductor. // Cells need to be in the map before join is called so it can route the call. self . conductor . write ( ) . await . add_cells ( cells ) ; // Join the network but ignore errors because the // space retries joining all cells every 5 minutes. futures :: stream :: iter ( networks ) . for_each_concurrent ( 100 , | mut network | async move { match tokio :: time :: timeout ( JOIN_NETWORK_TIMEOUT , network . join ( ) ) . await { Ok ( Err ( e ) ) => { tracing :: info ! ( failed_to_join_network e ) ; } Err ( _ ) => { tracing :: info ! ( " " ) ; } Ok ( Ok ( _ ) ) => ( ) , } } ) . await ; // Now we can trigger the workflows. for trigger in triggers { trigger . initialize_workflows ( ) ; } }
Add cells to the map then join the network then initialize workflows.
Add cells to the map then join the network then initialize workflows .
&self
cells
Vec<(Cell, InitialQueueTriggers)>
&self
cells
Vec<(Cell, InitialQueueTriggers)>