apt_swarm/
fetch.rs

1use crate::config;
2use crate::config::Repository;
3use crate::db::DatabaseClient;
4use crate::errors::*;
5use crate::keyring::Keyring;
6use crate::signed::Signed;
7use sequoia_openpgp::Fingerprint;
8use std::net::SocketAddr;
9use std::sync::Arc;
10use std::time::Duration;
11use tokio::task::JoinSet;
12
13pub const DEFAULT_CONCURRENCY: usize = 4;
14pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
15pub const READ_TIMEOUT: Duration = Duration::from_secs(60);
16
17async fn fetch_repository_updates(
18    client: &reqwest::Client,
19    keyring: &Option<Keyring>,
20    repository: &config::Repository,
21) -> Result<Vec<(Option<Fingerprint>, Signed)>> {
22    let mut out = Vec::new();
23
24    for source in &repository.urls {
25        let Ok(signed) = source
26            .fetch(client)
27            .await
28            .inspect_err(|err| error!("Error fetching latest release: {err:#}"))
29        else {
30            continue;
31        };
32
33        for item in signed.canonicalize(keyring.as_ref())? {
34            out.push(item);
35        }
36    }
37
38    Ok(out)
39}
40
41pub async fn fetch_updates<D: DatabaseClient>(
42    db: &mut D,
43    keyring: Arc<Option<Keyring>>,
44    concurrency: Option<usize>,
45    repositories: Vec<Repository>,
46    proxy: Option<SocketAddr>,
47) -> Result<()> {
48    let concurrency = concurrency.unwrap_or(DEFAULT_CONCURRENCY);
49    let mut queue = repositories.into_iter();
50    let mut pool = JoinSet::new();
51    let mut client = reqwest::Client::builder()
52        .connect_timeout(CONNECT_TIMEOUT)
53        .read_timeout(READ_TIMEOUT);
54    if let Some(proxy) = proxy {
55        let proxy = format!("socks5h://{proxy:?}");
56        let proxy = reqwest::Proxy::all(&proxy)
57            .with_context(|| anyhow!("Failed to parse as proxy: {proxy:?}"))?;
58        client = client.proxy(proxy);
59    }
60    let client = client.build().context("Failed to setup http client")?;
61
62    loop {
63        while pool.len() < concurrency {
64            if let Some(repository) = queue.next() {
65                let client = client.clone();
66                let keyring = keyring.clone();
67                pool.spawn(async move {
68                    fetch_repository_updates(&client, &keyring, &repository).await
69                });
70            } else {
71                // no more tasks to schedule
72                break;
73            }
74        }
75        if let Some(join) = pool.join_next().await {
76            match join.context("Failed to join task")? {
77                Ok(list) => {
78                    for (fp, variant) in list {
79                        let fp = fp.context(
80                            "Signature can't be imported because the signature is unverified",
81                        )?;
82                        db.add_release(&fp, &variant).await?;
83                    }
84                }
85                Err(err) => error!("Error fetching latest release: {err:#}"),
86            }
87        } else {
88            // no more tasks in pool
89            break;
90        }
91    }
92
93    Ok(())
94}
apt_swarm/fetch.rs

apt_swarm/
fetch.rs
