use reqwest client rather than global

This commit is contained in:
Igor Katson 2024-08-07 23:27:23 +01:00
parent 8c2788cf1a
commit 8c16239a0e
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
4 changed files with 29 additions and 4 deletions

View file

@ -196,12 +196,19 @@ pub struct Session {
default_storage_factory: Option<BoxStorageFactory>, default_storage_factory: Option<BoxStorageFactory>,
reqwest_client: reqwest::Client,
// This is stored for all tasks to stop when session is dropped. // This is stored for all tasks to stop when session is dropped.
_cancellation_token_drop_guard: DropGuard, _cancellation_token_drop_guard: DropGuard,
} }
async fn torrent_from_url(url: &str) -> anyhow::Result<TorrentMetaV1Owned> { async fn torrent_from_url(
let response = reqwest::get(url) reqwest_client: &reqwest::Client,
url: &str,
) -> anyhow::Result<TorrentMetaV1Owned> {
let response = reqwest_client
.get(url)
.send()
.await .await
.context("error downloading torrent metadata")?; .context("error downloading torrent metadata")?;
if !response.status().is_success() { if !response.status().is_success() {
@ -406,6 +413,11 @@ impl<'a> AddTorrent<'a> {
} }
} }
pub struct SocksProxyConfig {
// must start with socks5
pub url: String,
}
#[derive(Default)] #[derive(Default)]
pub struct SessionOptions { pub struct SessionOptions {
/// Turn on to disable DHT. /// Turn on to disable DHT.
@ -436,6 +448,8 @@ pub struct SessionOptions {
pub defer_writes_up_to: Option<usize>, pub defer_writes_up_to: Option<usize>,
pub default_storage_factory: Option<BoxStorageFactory>, pub default_storage_factory: Option<BoxStorageFactory>,
pub socks_proxy: Option<SocksProxyConfig>,
} }
async fn create_tcp_listener( async fn create_tcp_listener(
@ -534,6 +548,10 @@ impl Session {
}) })
.unwrap_or_default(); .unwrap_or_default();
let reqwest_client = reqwest::Client::builder()
.build()
.context("error building HTTP(S) client")?;
let session = Arc::new(Self { let session = Arc::new(Self {
persistence_filename, persistence_filename,
peer_id, peer_id,
@ -547,6 +565,7 @@ impl Session {
tcp_listen_port, tcp_listen_port,
disk_write_tx, disk_write_tx,
default_storage_factory: opts.default_storage_factory, default_storage_factory: opts.default_storage_factory,
reqwest_client,
}); });
if let Some(mut disk_write_rx) = disk_write_rx { if let Some(mut disk_write_rx) = disk_write_rx {
@ -922,7 +941,7 @@ impl Session {
AddTorrent::Url(url) AddTorrent::Url(url)
if url.starts_with("http://") || url.starts_with("https://") => if url.starts_with("http://") || url.starts_with("https://") =>
{ {
torrent_from_url(&url).await? torrent_from_url(&self.reqwest_client, &url).await?
} }
AddTorrent::Url(url) => { AddTorrent::Url(url) => {
bail!( bail!(
@ -1210,6 +1229,7 @@ impl Session {
Box::new(peer_rx_stats), Box::new(peer_rx_stats),
force_tracker_interval, force_tracker_interval,
announce_port, announce_port,
self.reqwest_client.clone(),
); );
Ok(merge_two_optional_streams(dht_rx, peer_rx)) Ok(merge_two_optional_streams(dht_rx, peer_rx))

View file

@ -74,6 +74,7 @@ async fn test_e2e() {
enable_upnp_port_forwarding: false, enable_upnp_port_forwarding: false,
default_storage_factory: None, default_storage_factory: None,
defer_writes_up_to: None, defer_writes_up_to: None,
..Default::default()
}, },
) )
.await .await

View file

@ -320,6 +320,7 @@ async fn async_main(opts: Opts) -> anyhow::Result<()> {
wrap(FilesystemStorageFactory::default()).boxed() wrap(FilesystemStorageFactory::default()).boxed()
} }
}), }),
socks_proxy: None,
}; };
let stats_printer = |session: Arc<Session>| async move { let stats_printer = |session: Arc<Session>| async move {

View file

@ -26,6 +26,7 @@ pub struct TrackerComms {
force_tracker_interval: Option<Duration>, force_tracker_interval: Option<Duration>,
tx: Sender, tx: Sender,
tcp_listen_port: Option<u16>, tcp_listen_port: Option<u16>,
reqwest_client: reqwest::Client,
} }
#[derive(Default)] #[derive(Default)]
@ -94,6 +95,7 @@ impl TrackerComms {
stats: Box<dyn TorrentStatsProvider>, stats: Box<dyn TorrentStatsProvider>,
force_interval: Option<Duration>, force_interval: Option<Duration>,
tcp_listen_port: Option<u16>, tcp_listen_port: Option<u16>,
reqwest_client: reqwest::Client,
) -> Option<BoxStream<'static, SocketAddr>> { ) -> Option<BoxStream<'static, SocketAddr>> {
let trackers = trackers let trackers = trackers
.into_iter() .into_iter()
@ -130,6 +132,7 @@ impl TrackerComms {
force_tracker_interval: force_interval, force_tracker_interval: force_interval,
tx, tx,
tcp_listen_port, tcp_listen_port,
reqwest_client
}); });
let mut futures = FuturesUnordered::new(); let mut futures = FuturesUnordered::new();
for tracker in trackers { for tracker in trackers {
@ -230,7 +233,7 @@ impl TrackerComms {
async fn tracker_one_request_http(&self, tracker_url: Url) -> anyhow::Result<u64> { async fn tracker_one_request_http(&self, tracker_url: Url) -> anyhow::Result<u64> {
debug!(url = ?tracker_url, "calling tracker over http"); debug!(url = ?tracker_url, "calling tracker over http");
let response: reqwest::Response = reqwest::get(tracker_url).await?; let response: reqwest::Response = self.reqwest_client.get(tracker_url).send().await?;
if !response.status().is_success() { if !response.status().is_success() {
anyhow::bail!("tracker responded with {:?}", response.status()); anyhow::bail!("tracker responded with {:?}", response.status());
} }