pub async fn wait_for_token<F>(write_token: impl Into<String>, connection: Connection, f: F)
where
    F: Fn(&GetWriteInfoResponse) -> bool,
{
    let write_token = write_token.into();
    assert!(!write_token.is_empty());

    info!("  write token: {}", write_token);

    let retry_duration = Duration::from_secs(MAX_QUERY_RETRY_TIME_SEC);
    let mut write_info_client = influxdb_iox_client::write_info::Client::new(connection);
    tokio::time::timeout(retry_duration, async move {
        let mut interval = tokio::time::interval(Duration::from_millis(500));
        loop {
            match write_info_client.get_write_info(&write_token).await {
                Ok(res) => {
                    if f(&res) {
                        return;
                    }
                    info!("Retrying; predicate not satistified: {:?}", res);
                }

                Err(e) => {
                    info!("Retrying; Got error getting write_info: {}", e);
                }
            };
            interval.tick().await;
        }
    })
    .await
    .expect("did not get passing predicate on token");
}
/// Waits for the specified predicate to return true
Waits for the specified predicate to return true
Waits for the specified predicate to return true
pub async fn wait_for_token<F>(write_token: impl Into<String>, connection: Connection, f: F)
where
    F: Fn(&GetWriteInfoResponse) -> bool,
{
    let write_token = write_token.into();
    assert!(!write_token.is_empty());
    info!("  write token: {}", write_token);
    let retry_duration = Duration::from_secs(MAX_QUERY_RETRY_TIME_SEC);
    let mut write_info_client = influxdb_iox_client::write_info::Client::new(connection);
    tokio::time::timeout(retry_duration, async move {
        let mut interval = tokio::time::interval(Duration::from_millis(500));
        loop {
            match write_info_client.get_write_info(&write_token).await {
                Ok(res) => {
                    if f(&res) {
                        return;
                    }
                    info!("Retrying; predicate not satistified: {:?}", res);
                }
                Err(e) => {
                    info!("Retrying; Got error getting write_info: {}", e);
                }
            };
            interval.tick().await;
        }
    })
    .await
    .expect("did not get passing predicate on token");
}
pub async fn wait_for_token < F > ( write_token : impl Into < String > , connection : Connection , f : F ) where F : Fn ( & GetWriteInfoResponse ) -> bool , { let write_token = write_token . into ( ) ; assert ! ( write_token is_empty ( ) ) ; info ! ( " " write_token ) ; let retry_duration = Duration :: from_secs ( MAX_QUERY_RETRY_TIME_SEC ) ; let mut write_info_client = influxdb_iox_client :: write_info :: Client :: new ( connection ) ; tokio :: time :: timeout ( retry_duration , async move { let mut interval = tokio :: time :: interval ( Duration :: from_millis ( 500 ) ) ; loop { match write_info_client . get_write_info ( & write_token ) . await { Ok ( res ) => { if f ( & res ) { return ; } info ! ( " " res ) ; } Err ( e ) => { info ! ( " " e ) ; } } ; interval . tick ( ) . await ; } } ) . await . expect ( " " ) ; }
Waits for the specified predicate to return true
Waits for the specified predicate to return true
write_token
impl Into<String>
connection
Connection
f
F
write_token
impl Into<String>
connection
Connection
f
F