pub(crate) async fn run(config: &Config, db: &impl GenericClient) -> Result<()> {
    // Some duration to wait before the next attempt. Is only set to non-zero in
    // case of an error.
    let mut backoff = INITIAL_BACKOFF;

    /// Helper macro to call in case of not being able to get a proper response
    /// from Opencast. Forwards all arguments to `error!`, increases `backoff`
    /// and sleeps for the backoff period.
    macro_rules! request_failed {
        ($($t:tt)*) => {{
            error!($($t)*);

            // We increase the backoff duration exponentially until we hit the
            // defined maximum.
            info!("Waiting {:.1?} due to error before trying again", backoff);
            tokio::time::sleep(backoff).await;
            backoff = min(MAX_BACKOFF, backoff.mul_f32(1.5));

            continue;
        }};
    }

    let client = HarvestClient::new(config);

    loop {
        let sync_status = SyncStatus::fetch(db).await
            .context("failed to fetch sync status from DB")?;


        // Send request to API and deserialize data.
        let (response, body) = match client.send(sync_status.harvested_until, HARVEST_LIMIT).await {
            Ok(v) => v,
            Err(e) => request_failed!("Harvest request failed: {:?}", e),
        };

        if response.status != StatusCode::OK {
            trace!("HTTP response: {:#?}", response);
            request_failed!("Harvest API returned unexepcted HTTP code {}", response.status);
        }

        let harvest_data = match serde_json::from_slice::<HarvestResponse>(&body) {
            Ok(v) => v,
            Err(e) => {
                trace!("HTTP response: {:#?}", response);
                request_failed!("Failed to deserialize response from harvesting API: {}", e);
            }
        };

        // Communication with Opencast succeeded: reset backoff time.
        backoff = INITIAL_BACKOFF;


        // Write received data into the database, updating the sync status if
        // everything worked out alright.
        store_in_db(&harvest_data.items, &sync_status, db).await?;
        SyncStatus::update_harvested_until(harvest_data.includes_items_until, db).await?;
        if !harvest_data.has_more {
            debug!(
                "Harvested all available data: waiting {:?} before starting next harvest",
                POLL_PERIOD,
            );

            tokio::time::sleep(POLL_PERIOD).await;
        }
    }
}
/// Continuiously fetches from the harvesting API and writes new data into our
/// database.
Continuiously fetches from the harvesting API and writes new data into our
database.
Continuiously fetches from the harvesting API and writes new data into our database .
pub(crate) async fn run(config: &Config, db: &impl GenericClient) -> Result<()> {
    // Some duration to wait before the next attempt. Is only set to non-zero in
    // case of an error.
    let mut backoff = INITIAL_BACKOFF;
    /// Helper macro to call in case of not being able to get a proper response
    /// from Opencast. Forwards all arguments to `error!`, increases `backoff`
    /// and sleeps for the backoff period.
    macro_rules! request_failed {
        ($($t:tt)*) => {{
            error!($($t)*);
            // We increase the backoff duration exponentially until we hit the
            // defined maximum.
            info!("Waiting {:.1?} due to error before trying again", backoff);
            tokio::time::sleep(backoff).await;
            backoff = min(MAX_BACKOFF, backoff.mul_f32(1.5));
            continue;
        }};
    }
    let client = HarvestClient::new(config);
    loop {
        let sync_status = SyncStatus::fetch(db).await
            .context("failed to fetch sync status from DB")?;
        // Send request to API and deserialize data.
        let (response, body) = match client.send(sync_status.harvested_until, HARVEST_LIMIT).await {
            Ok(v) => v,
            Err(e) => request_failed!("Harvest request failed: {:?}", e),
        };
        if response.status != StatusCode::OK {
            trace!("HTTP response: {:#?}", response);
            request_failed!("Harvest API returned unexepcted HTTP code {}", response.status);
        }
        let harvest_data = match serde_json::from_slice::<HarvestResponse>(&body) {
            Ok(v) => v,
            Err(e) => {
                trace!("HTTP response: {:#?}", response);
                request_failed!("Failed to deserialize response from harvesting API: {}", e);
            }
        };
        // Communication with Opencast succeeded: reset backoff time.
        backoff = INITIAL_BACKOFF;
        // Write received data into the database, updating the sync status if
        // everything worked out alright.
        store_in_db(&harvest_data.items, &sync_status, db).await?;
        SyncStatus::update_harvested_until(harvest_data.includes_items_until, db).await?;
        if !harvest_data.has_more {
            debug!(
                "Harvested all available data: waiting {:?} before starting next harvest",
                POLL_PERIOD,
            );
            tokio::time::sleep(POLL_PERIOD).await;
        }
    }
}
pub ( crate ) async fn run ( config : & Config , db : & impl GenericClient ) -> Result < ( ) > { // Some duration to wait before the next attempt. Is only set to non-zero in // case of an error. let mut backoff = INITIAL_BACKOFF ; /// Helper macro to call in case of not being able to get a proper response /// from Opencast. Forwards all arguments to `error!`, increases `backoff` /// and sleeps for the backoff period. macro_rules! request_failed { ( $ ( $t : tt ) * ) => { { error ( $ ( $t ) * ) // We increase the backoff duration exponentially until we hit the // defined maximum. info ( " " backoff ) tokio time sleep ( backoff ) await backoff min ( MAX_BACKOFF backoff mul_f32 ( 1.5 ) ) continue } } ; } let client = HarvestClient :: new ( config ) ; loop { let sync_status = SyncStatus :: fetch ( db ) . await . context ( " " ) ? ; // Send request to API and deserialize data. let ( response , body ) = match client . send ( sync_status . harvested_until , HARVEST_LIMIT ) . await { Ok ( v ) => v , Err ( e ) => request_failed ! ( " " e ) , } ; if response . status != StatusCode :: OK { trace ! ( " " response ) ; request_failed ! ( " " response status ) ; } let harvest_data = match serde_json :: from_slice :: < HarvestResponse > ( & body ) { Ok ( v ) => v , Err ( e ) => { trace ! ( " " response ) ; request_failed ! ( " " e ) ; } } ; // Communication with Opencast succeeded: reset backoff time. backoff = INITIAL_BACKOFF ; // Write received data into the database, updating the sync status if // everything worked out alright. store_in_db ( & harvest_data . items , & sync_status , db ) . await ? ; SyncStatus :: update_harvested_until ( harvest_data . includes_items_until , db ) . await ? ; if ! harvest_data . has_more { debug ! ( " " POLL_PERIOD ) ; tokio :: time :: sleep ( POLL_PERIOD ) . await ; } } }
Continuiously fetches from the harvesting API and writes new data into our
database.
Continuiously fetches from the harvesting API and writes new data into our database .
config
&Config
db
&impl GenericClient