Function prefetch_dl_task

opensuse-proxy-cache/src/main.rs:1122–1172

(
    client: reqwest::Client,
    fetch_url: Url,
    submit_tx: Sender<CacheMeta>,
    cache_key: PathBuf,
    tmp_file: NamedTempFile,
    cls: Classification,
)

#[instrument(skip_all)]
async fn prefetch_dl_task(
    client: reqwest::Client,
    fetch_url: Url,
    submit_tx: Sender<CacheMeta>,
    cache_key: PathBuf,
    tmp_file: NamedTempFile,
    cls: Classification,
) {
    info!("🚅  start prefetch {}", cache_key.display());

    let send_headers = send_headers(None);

    let client_response = client.get(fetch_url).headers(send_headers).send().await;

    let client_response = match client_response {
        Ok(cr) => cr,
        Err(e) => {
            error!(?e, "Error handling client response");
            return;
        }
    };

    let status = client_response.status();
    if status == StatusCode::NOT_FOUND {
        info!("👻  prefetch rewrite -> NotFound");
        let etime = time::OffsetDateTime::now_utc();
        let _ = submit_tx
            .send(CacheMeta {
                cache_key,
                etime,
                action: Action::NotFound { cls },
            })
            .await;
        return;
    } else if status != StatusCode::OK {
        error!("Response returned {:?}, aborting prefetch", status);
        return;
    }

    let headers = filter_headers(client_response.headers(), false);

    let byte_reader = StreamReader::new(
        client_response
            .bytes_stream()
            .map(|item| item.map_err(|e| std::io::Error::other(e))),
    );

    let _ = tokio::task::spawn(async move {
        write_file(cache_key, headers, tmp_file, submit_tx, cls, byte_reader).await
    });
}

#[instrument(skip_all)]
async fn found(