async fn connect(&self) -> Result<Connection, Error> {
        let mut maybe_connection = self.maybe_connection.lock().await;

        let ingester_address = self.ingester_address.as_ref();

        if let Some(connection) = maybe_connection.as_ref() {
            debug!(%ingester_address, "Reusing connection to ingester");

            Ok(connection.clone())
        } else {
            debug!(%ingester_address, "Connecting to ingester");

            let connection = connection::Builder::new()
                .build(ingester_address)
                .await
                .context(ConnectingSnafu { ingester_address })?;

            // sanity check w/ a handshake
            let mut client = flight::Client::<flight::generated_types::IngesterQueryRequest>::new(
                connection.clone(),
            );

            // make contact with the ingester
            client
                .handshake()
                .await
                .context(HandshakeSnafu { ingester_address })?;

            *maybe_connection = Some(connection.clone());
            Ok(connection)
        }
    }
/// Return the underlying connection, creating it if needed
Return the underlying connection, creating it if needed
Return the underlying connection creating it if needed
async fn connect(&self) -> Result<Connection, Error> {
        let mut maybe_connection = self.maybe_connection.lock().await;
        let ingester_address = self.ingester_address.as_ref();
        if let Some(connection) = maybe_connection.as_ref() {
            debug!(%ingester_address, "Reusing connection to ingester");
            Ok(connection.clone())
        } else {
            debug!(%ingester_address, "Connecting to ingester");
            let connection = connection::Builder::new()
                .build(ingester_address)
                .await
                .context(ConnectingSnafu { ingester_address })?;
            // sanity check w/ a handshake
            let mut client = flight::Client::<flight::generated_types::IngesterQueryRequest>::new(
                connection.clone(),
            );
            // make contact with the ingester
            client
                .handshake()
                .await
                .context(HandshakeSnafu { ingester_address })?;
            *maybe_connection = Some(connection.clone());
            Ok(connection)
        }
    }
async fn connect ( & self ) -> Result < Connection , Error > { let mut maybe_connection = self . maybe_connection . lock ( ) . await ; let ingester_address = self . ingester_address . as_ref ( ) ; if let Some ( connection ) = maybe_connection . as_ref ( ) { debug ! ( ingester_address " " ) ; Ok ( connection . clone ( ) ) } else { debug ! ( ingester_address " " ) ; let connection = connection :: Builder :: new ( ) . build ( ingester_address ) . await . context ( ConnectingSnafu { ingester_address } ) ? ; // sanity check w/ a handshake let mut client = flight :: Client :: < flight :: generated_types :: IngesterQueryRequest > :: new ( connection . clone ( ) , ) ; // make contact with the ingester client . handshake ( ) . await . context ( HandshakeSnafu { ingester_address } ) ? ; * maybe_connection = Some ( connection . clone ( ) ) ; Ok ( connection ) } }
Return the underlying connection, creating it if needed
Return the underlying connection creating it if needed
&self
&self