Function prefetch_task

opensuse-proxy-cache/src/main.rs:1411–1466

(
    state: Arc<AppState>,
    mut prefetch_rx: Receiver<PrefetchReq>,
    mut rx: broadcast::Receiver<bool>,
)

}

async fn prefetch_task(
    state: Arc<AppState>,
    mut prefetch_rx: Receiver<PrefetchReq>,
    mut rx: broadcast::Receiver<bool>,
) {
    info!(immediate = true, "Spawning prefetch task ...");

    let mut req_cache = LruCache::new(NonZeroUsize::new(64).unwrap());

    loop {
        tokio::select! {
            _ = rx.recv() => { break; }
            _ = sleep(Duration::from_secs(300)) => {
                // Do nothing, this is to make us loop and check the running state.
                debug!("prefetch loop idle");
            }
            got = prefetch_rx.recv() => {
                async {
                    match got {
                        Some(PrefetchReq {
                            cache_key,
                            fetch_url,
                            tmp_file,
                            submit_tx,
                            cls
                        }) => {
                            trace!("received a prefetch operation");
                            let debounce_t = req_cache.get(&cache_key)
                                .map(|inst: &Instant| inst.elapsed().as_secs())
                                .unwrap_or(DEBOUNCE + 1);
                            let debounce = debounce_t < DEBOUNCE;

                            if debounce {
                                debug!(immediate = true, "Skipping debounce item {}", cache_key.display());
                            } else {
                                // Sometimes if the dl is large, we can accidentally trigger a second dl because the cache
                                // hasn't finished crc32c yet. So we need a tiny cache to debounce repeat dl's.
                                req_cache.put(cache_key.clone(), Instant::now());
                                prefetch_dl_task(state.client.clone(), fetch_url, submit_tx, cache_key, tmp_file, cls).await;
                            }
                        }
                        None => {
                            // channels dead.
                            warn!("prefetch channel has died");
                            // return;
                        }
                    }
                }
                .instrument(tracing::info_span!("prefetch_task"))
                .await;
            }
        }
    }

    warn!(immediate = true, "Stopping prefetch task.");
}

async fn ipxe_static(extract::Path(fname): extract::Path<PathBuf>) -> Response {