async fn wait_until_ready(&self) -> Connections {
        let mut need_wait_for_startup = false;
        {
            let mut ready = self.ready.lock().await;
            match *ready {
                ServerState::Started => {
                    // first time, need to try and start it
                    need_wait_for_startup = true;
                    *ready = ServerState::Starting;
                }
                ServerState::Starting => {
                    // someone else is starting this
                }
                ServerState::Ready => {}
                ServerState::Error => {
                    panic!("Server was previously found to be in Error, aborting");
                }
            };
        }

        // at first, attempt to reconnect all the clients
        let mut connections = Connections::new();
        async {
            let mut interval = tokio::time::interval(Duration::from_millis(100));
            loop {
                match connections.reconnect(&self.test_config).await {
                    Err(e) => info!("wait_until_ready: can not yet connect: {}", e),
                    Ok(()) => return,
                }
                interval.tick().await;
            }
        }
        .with_timeout_panic(Duration::from_secs(10))
        .await;

        if !need_wait_for_startup {
            return connections;
        }

        // Poll the RPC and HTTP servers separately as they listen on
        // different ports but both need to be up for the test to run
        let try_grpc_connect = self.wait_for_grpc(&connections);

        let server_process = Arc::clone(&self.server_process);
        let try_http_connect = async {
            let client = reqwest::Client::new();
            let url = format!("{}/health", self.addrs().router_http_api().client_base());
            let mut interval = tokio::time::interval(Duration::from_millis(1000));
            loop {
                if server_dead(server_process.as_ref()).await {
                    break;
                }
                match client.get(&url).send().await {
                    Ok(resp) => {
                        info!(
                            "Successfully got a response from {:?} HTTP: {:?}",
                            self.test_config.server_type(),
                            resp
                        );
                        return;
                    }
                    Err(e) => {
                        info!(
                            "Waiting for {:?} HTTP server to be up: {}",
                            self.test_config.server_type(),
                            e
                        );
                    }
                }
                interval.tick().await;
            }
        };

        let pair = future::join(try_http_connect, try_grpc_connect);

        let capped_check = tokio::time::timeout(Duration::from_secs(30), pair);

        match capped_check.await {
            Ok(_) => {
                info!("Successfully started {}", self);
                let mut ready = self.ready.lock().await;
                *ready = ServerState::Ready;
            }
            Err(e) => {
                // tell others that this server had some problem
                let mut ready = self.ready.lock().await;
                *ready = ServerState::Error;
                std::mem::drop(ready);
                panic!(
                    "{:?} Server was not ready in required time: {}",
                    self.test_config.server_type(),
                    e
                );
            }
        }

        connections
    }
/// Polls the various services to ensure the server is
/// operational, reestablishing grpc connections, and returning
/// those active connections
Polls the various services to ensure the server is
operational, reestablishing grpc connections, and returning
those active connections
Polls the various services to ensure the server is operational reestablishing grpc connections and returning those active connections
async fn wait_until_ready(&self) -> Connections {
        let mut need_wait_for_startup = false;
        {
            let mut ready = self.ready.lock().await;
            match *ready {
                ServerState::Started => {
                    // first time, need to try and start it
                    need_wait_for_startup = true;
                    *ready = ServerState::Starting;
                }
                ServerState::Starting => {
                    // someone else is starting this
                }
                ServerState::Ready => {}
                ServerState::Error => {
                    panic!("Server was previously found to be in Error, aborting");
                }
            };
        }
        // at first, attempt to reconnect all the clients
        let mut connections = Connections::new();
        async {
            let mut interval = tokio::time::interval(Duration::from_millis(100));
            loop {
                match connections.reconnect(&self.test_config).await {
                    Err(e) => info!("wait_until_ready: can not yet connect: {}", e),
                    Ok(()) => return,
                }
                interval.tick().await;
            }
        }
        .with_timeout_panic(Duration::from_secs(10))
        .await;
        if !need_wait_for_startup {
            return connections;
        }
        // Poll the RPC and HTTP servers separately as they listen on
        // different ports but both need to be up for the test to run
        let try_grpc_connect = self.wait_for_grpc(&connections);
        let server_process = Arc::clone(&self.server_process);
        let try_http_connect = async {
            let client = reqwest::Client::new();
            let url = format!("{}/health", self.addrs().router_http_api().client_base());
            let mut interval = tokio::time::interval(Duration::from_millis(1000));
            loop {
                if server_dead(server_process.as_ref()).await {
                    break;
                }
                match client.get(&url).send().await {
                    Ok(resp) => {
                        info!(
                            "Successfully got a response from {:?} HTTP: {:?}",
                            self.test_config.server_type(),
                            resp
                        );
                        return;
                    }
                    Err(e) => {
                        info!(
                            "Waiting for {:?} HTTP server to be up: {}",
                            self.test_config.server_type(),
                            e
                        );
                    }
                }
                interval.tick().await;
            }
        };
        let pair = future::join(try_http_connect, try_grpc_connect);
        let capped_check = tokio::time::timeout(Duration::from_secs(30), pair);
        match capped_check.await {
            Ok(_) => {
                info!("Successfully started {}", self);
                let mut ready = self.ready.lock().await;
                *ready = ServerState::Ready;
            }
            Err(e) => {
                // tell others that this server had some problem
                let mut ready = self.ready.lock().await;
                *ready = ServerState::Error;
                std::mem::drop(ready);
                panic!(
                    "{:?} Server was not ready in required time: {}",
                    self.test_config.server_type(),
                    e
                );
            }
        }
        connections
    }
async fn wait_until_ready ( & self ) -> Connections { let mut need_wait_for_startup = false ; { let mut ready = self . ready . lock ( ) . await ; match * ready { ServerState :: Started => { // first time, need to try and start it need_wait_for_startup = true ; * ready = ServerState :: Starting ; } ServerState :: Starting => { // someone else is starting this } ServerState :: Ready => { } ServerState :: Error => { panic ! ( " " ) ; } } ; } // at first, attempt to reconnect all the clients let mut connections = Connections :: new ( ) ; async { let mut interval = tokio :: time :: interval ( Duration :: from_millis ( 100 ) ) ; loop { match connections . reconnect ( & self . test_config ) . await { Err ( e ) => info ! ( " " e ) , Ok ( ( ) ) => return , } interval . tick ( ) . await ; } } . with_timeout_panic ( Duration :: from_secs ( 10 ) ) . await ; if ! need_wait_for_startup { return connections ; } // Poll the RPC and HTTP servers separately as they listen on // different ports but both need to be up for the test to run let try_grpc_connect = self . wait_for_grpc ( & connections ) ; let server_process = Arc :: clone ( & self . server_process ) ; let try_http_connect = async { let client = reqwest :: Client :: new ( ) ; let url = format ! ( " " self addrs ( ) router_http_api ( ) client_base ( ) ) ; let mut interval = tokio :: time :: interval ( Duration :: from_millis ( 1000 ) ) ; loop { if server_dead ( server_process . as_ref ( ) ) . await { break ; } match client . get ( & url ) . send ( ) . await { Ok ( resp ) => { info ! ( " " self test_config server_type ( ) resp ) ; return ; } Err ( e ) => { info ! ( " " self test_config server_type ( ) e ) ; } } interval . tick ( ) . await ; } } ; let pair = future :: join ( try_http_connect , try_grpc_connect ) ; let capped_check = tokio :: time :: timeout ( Duration :: from_secs ( 30 ) , pair ) ; match capped_check . await { Ok ( _ ) => { info ! ( " " self ) ; let mut ready = self . ready . lock ( ) . await ; * ready = ServerState :: Ready ; } Err ( e ) => { // tell others that this server had some problem let mut ready = self . ready . lock ( ) . await ; * ready = ServerState :: Error ; std :: mem :: drop ( ready ) ; panic ! ( " " self test_config server_type ( ) e ) ; } } connections }
Polls the various services to ensure the server is
operational, reestablishing grpc connections, and returning
those active connections
Polls the various services to ensure the server is operational reestablishing grpc connections and returning those active connections
&self
&self