use std::net::IpAddr;
use std::sync::Arc;
use super::{Headers, Response};
use crate::BoxFuture;
#[derive(Clone, Debug)]
pub struct FetchInit {
pub method: String,
pub headers: Headers,
pub body: Option<Vec<u8>>,
pub manual_redirect: bool,
pub timeout_ms: u64,
pub max_bytes: Option<usize>,
pub truncate: bool,
pub resolve: Vec<(String, u16, Vec<IpAddr>)>,
}
impl Default for FetchInit {
fn default() -> Self {
FetchInit {
method: "GET".into(),
headers: Headers::new(),
body: None,
manual_redirect: false,
timeout_ms: 30_000,
max_bytes: None,
truncate: false,
resolve: Vec::new(),
}
}
}
impl FetchInit {
pub fn method(method: &str) -> FetchInit {
FetchInit { method: method.to_ascii_uppercase(), ..FetchInit::default() }
}
pub fn header(mut self, name: &str, value: impl AsRef<str>) -> FetchInit {
self.headers.set(name, value.as_ref());
self
}
pub fn body(mut self, body: impl Into<Vec<u8>>) -> FetchInit {
self.body = Some(body.into());
self
}
pub fn timeout(mut self, ms: u64) -> FetchInit {
self.timeout_ms = ms;
self
}
pub fn max_bytes(mut self, max: usize) -> FetchInit {
self.max_bytes = Some(max);
self
}
}
pub const DEFAULT_MAX_BYTES: usize = 100 * 1024 * 1024;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FetchError {
Failed(String),
TimedOut,
TooLong(usize),
}
impl std::fmt::Display for FetchError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
FetchError::Failed(m) => f.write_str(m),
FetchError::TimedOut => f.write_str("The operation was aborted due to timeout"),
FetchError::TooLong(n) => write!(f, "Body over {n} bytes"),
}
}
}
impl std::error::Error for FetchError {}
pub trait Fetcher: Send + Sync {
fn fetch<'a>(&'a self, url: &'a str, init: FetchInit) -> BoxFuture<'a, Result<Response, FetchError>>;
fn pins(&self) -> bool {
false
}
}
pub type SharedFetcher = Arc<dyn Fetcher>;
#[cfg(feature = "transport")]
#[derive(Clone)]
pub struct ReqwestFetcher {
follow: reqwest::Client,
manual: reqwest::Client,
}
#[cfg(feature = "transport")]
impl Default for ReqwestFetcher {
fn default() -> Self {
ReqwestFetcher::new()
}
}
#[cfg(feature = "transport")]
impl ReqwestFetcher {
pub fn new() -> ReqwestFetcher {
ReqwestFetcher { follow: client(false, &[]).unwrap_or_default(), manual: client(true, &[]).unwrap_or_default() }
}
}
#[cfg(feature = "transport")]
fn client(manual: bool, resolve: &[(String, u16, Vec<IpAddr>)]) -> Result<reqwest::Client, FetchError> {
let tls = crate::mail::smtp::tls_config().map_err(FetchError::Failed)?;
let mut b = reqwest::Client::builder()
.tls_backend_preconfigured((*tls).clone())
.redirect(if manual { reqwest::redirect::Policy::none() } else { reqwest::redirect::Policy::limited(20) })
.connect_timeout(std::time::Duration::from_secs(15));
if !resolve.is_empty() {
b = b.no_proxy();
}
for (host, port, addresses) in resolve {
if addresses.is_empty() {
return Err(FetchError::Failed(format!("{host} has no address to connect to")));
}
let addrs: Vec<std::net::SocketAddr> =
addresses.iter().map(|ip| std::net::SocketAddr::new(*ip, *port)).collect();
b = b.resolve_to_addrs(host, &addrs);
}
b.build().map_err(|e| FetchError::Failed(e.to_string()))
}
#[cfg(feature = "transport")]
impl Fetcher for ReqwestFetcher {
fn pins(&self) -> bool {
true
}
fn fetch<'a>(&'a self, url: &'a str, init: FetchInit) -> BoxFuture<'a, Result<Response, FetchError>> {
Box::pin(async move {
let pinned;
let http = if !init.resolve.is_empty() {
pinned = client(init.manual_redirect, &init.resolve)?;
&pinned
} else if init.manual_redirect {
&self.manual
} else {
&self.follow
};
let method =
reqwest::Method::from_bytes(init.method.as_bytes()).map_err(|e| FetchError::Failed(e.to_string()))?;
let mut req = http.request(method, url).timeout(std::time::Duration::from_millis(init.timeout_ms.max(1)));
for (name, values) in init.headers.all() {
for v in values {
req = req.header(name.as_str(), v.as_str());
}
}
if let Some(body) = init.body {
req = req.body(body);
}
let fail = |e: reqwest::Error| {
if e.is_timeout() { FetchError::TimedOut } else { FetchError::Failed(e.to_string()) }
};
let mut answer = req.send().await.map_err(fail)?;
let status = answer.status().as_u16();
let mut headers = Headers::new();
for (name, value) in answer.headers() {
headers.append(name.as_str(), &String::from_utf8_lossy(value.as_bytes()));
}
let max = init.max_bytes.unwrap_or(DEFAULT_MAX_BYTES);
if !init.truncate && answer.content_length().is_some_and(|n| n > max as u64) {
return Err(FetchError::TooLong(max));
}
let mut body = Vec::new();
loop {
match answer.chunk().await {
Ok(Some(chunk)) => {
if body.len() + chunk.len() > max {
if init.truncate {
body.extend_from_slice(&chunk[..max - body.len()]);
break;
}
return Err(FetchError::TooLong(max));
}
body.extend_from_slice(&chunk);
}
Ok(None) => break,
Err(e) => return Err(fail(e)),
}
}
Ok(Response { status, headers, body })
})
}
}
pub fn default_fetcher() -> SharedFetcher {
#[cfg(feature = "transport")]
{
Arc::new(ReqwestFetcher::new())
}
#[cfg(not(feature = "transport"))]
{
Arc::new(NoFetcher)
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct NoFetcher;
impl Fetcher for NoFetcher {
fn fetch<'a>(&'a self, _url: &'a str, _init: FetchInit) -> BoxFuture<'a, Result<Response, FetchError>> {
Box::pin(async { Err(FetchError::Failed("fetch failed: Runlight was built without a transport".into())) })
}
}