use std::{
io::{self, Read, Seek, SeekFrom},
sync::atomic::{AtomicBool, Ordering},
time::Duration,
};
use anyhow::{Context, Result, anyhow};
use reqwest::{
blocking::{Client, Response},
header,
};
use url::Url;
use crate::utils::format_size;
static ACCEPT_RANGES_WARNING_SHOWN: AtomicBool = AtomicBool::new(false);
static FILE_SIZE_INFO_SHOWN: AtomicBool = AtomicBool::new(false);
#[derive(Clone)]
pub struct HttpReader {
url: String,
position: u64,
pub content_length: u64,
client: Client,
pub content_type: Option<String>,
}
impl HttpReader {
pub fn new(url: String, user_agent: &str) -> Result<Self> {
Self::new_internal(url, true, user_agent)
}
pub fn new_silent(url: String, user_agent: &str) -> Result<Self> {
Self::new_internal(url, false, user_agent)
}
fn create_client(user_agent: &str) -> Result<Client> {
let mut headers = header::HeaderMap::new();
headers.insert(
header::ACCEPT_ENCODING,
header::HeaderValue::from_static("gzip, deflate, br"),
);
headers.insert(header::ACCEPT, header::HeaderValue::from_static("*/*"));
headers.insert(
header::USER_AGENT,
header::HeaderValue::from_str(user_agent)?,
);
headers.insert(
header::ACCEPT_RANGES,
header::HeaderValue::from_static("bytes"),
);
headers.insert(
header::CONNECTION,
header::HeaderValue::from_static("keep-alive"),
);
headers.insert(
header::CACHE_CONTROL,
header::HeaderValue::from_static("no-transform"),
);
let client_builder = Client::builder()
.timeout(Duration::from_secs(600))
.tcp_keepalive(Some(Duration::from_secs(30)))
.pool_max_idle_per_host(10)
.default_headers(headers)
.redirect(reqwest::redirect::Policy::limited(10));
let client = client_builder.build()?;
Ok(client)
}
fn new_internal(url: String, print_size: bool, user_agent: &str) -> Result<Self> {
let client = Self::create_client(user_agent).with_context(|| "Failed to build client")?;
let parsed_url = Url::parse(&url).map_err(|e| anyhow!("Invalid URL: {e}"))?;
let _host = parsed_url
.host_str()
.ok_or_else(|| anyhow!("No host in URL"))?;
let _port = parsed_url.port().unwrap_or_else(|| {
if parsed_url.scheme() == "https" {
443
} else {
80
}
});
let mut retry_count: u64 = 0;
let max_retries = 3;
let mut last_error = None;
while retry_count < max_retries {
match client.head(&url).send() {
Ok(response) => {
let content_type = response
.headers()
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(std::string::ToString::to_string);
let content_length = response
.headers()
.get(header::CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse::<u64>().ok())
.ok_or_else(|| anyhow!("Could not determine content length"))?;
let supports_ranges = response
.headers()
.get(header::ACCEPT_RANGES)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v == "bytes");
if !supports_ranges && !ACCEPT_RANGES_WARNING_SHOWN.swap(true, Ordering::SeqCst)
{
eprintln!(
"- Warning: Server doesn't advertise Accept-Ranges. The process may fail."
);
}
if print_size && !FILE_SIZE_INFO_SHOWN.swap(true, Ordering::SeqCst) {
eprintln!("- File size: {}", format_size(content_length));
}
return Ok(Self {
url,
position: 0,
content_length,
client,
content_type,
});
}
Err(e) => {
last_error = Some(e);
retry_count += 1;
if retry_count < max_retries {
std::thread::sleep(Duration::from_secs(2 * retry_count));
}
}
}
}
Err(anyhow!(
"Failed to connect after {max_retries} retries. Last error: {}",
last_error.unwrap()
))
}
pub fn read_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
if offset >= self.content_length {
return Ok(0);
}
let remaining = self.content_length - offset;
let to_read = std::cmp::min(buf.len(), remaining as usize) as usize;
if to_read == 0 {
return Ok(0);
}
let end = offset + to_read as u64 - 1;
let range = format!("bytes={offset}-{end}");
let mut retry_count = 0;
let max_retries = 3;
while retry_count < max_retries {
match self
.client
.get(&self.url)
.header(header::RANGE, range.clone())
.send()
{
Ok(mut response) => {
if !response.status().is_success() && response.status().as_u16() != 206 {
return Err(io::Error::other(format!(
"HTTP error: {} for range {}",
response.status(),
range
)));
}
return copy_from_response(&mut response, &mut buf[..to_read]);
}
Err(e) => {
retry_count += 1;
if retry_count == max_retries {
return Err(io::Error::other(format!(
"Failed to read range {range} after {max_retries} retries: {e}"
)));
}
std::thread::sleep(Duration::from_secs(2 * retry_count as u64));
}
}
}
Err(io::Error::other(
"Failed to read range after maximum retries",
))
}
}
impl Read for HttpReader {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let bytes_read = self.read_at(self.position, buf)?;
self.position += bytes_read as u64;
Ok(bytes_read)
}
}
impl Seek for HttpReader {
fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
let new_pos = match pos {
SeekFrom::Start(offset) => offset,
SeekFrom::Current(offset) => {
if offset >= 0 {
self.position.saturating_add(offset as u64)
} else {
self.position.saturating_sub(offset.unsigned_abs())
}
}
SeekFrom::End(offset) => {
if offset >= 0 {
self.content_length.saturating_add(offset as u64)
} else {
self.content_length.saturating_sub(offset.unsigned_abs())
}
}
};
if new_pos > self.content_length {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"Seek position {} exceeds content length {}",
new_pos, self.content_length
),
));
}
self.position = new_pos;
Ok(self.position)
}
}
pub fn copy_from_response(response: &mut Response, buf: &mut [u8]) -> io::Result<usize> {
let mut total_read = 0;
while total_read < buf.len() {
match response.read(&mut buf[total_read..]) {
Ok(0) => break, Ok(n) => total_read += n,
Err(e) if e.kind() == io::ErrorKind::Interrupted => (),
Err(e) => return Err(e),
}
}
Ok(total_read)
}