use std::io::Read; use std::{any::Any, convert::TryFrom, io::Write, str::FromStr}; use git_features::io::pipe; use crate::client::http::traits::PostBodyDataKind; use crate::client::{http, http::reqwest::Remote}; /// The error returned by the 'remote' helper, a purely internal construct to perform http requests. #[derive(Debug, thiserror::Error)] #[allow(missing_docs)] pub enum Error { #[error(transparent)] Reqwest(#[from] reqwest::Error), #[error("Could not finish reading all data to post to the remote")] ReadPostBody(#[from] std::io::Error), #[error("Request configuration failed")] ConfigureRequest(#[from] Box), } impl crate::IsSpuriousError for Error { fn is_spurious(&self) -> bool { match self { Error::Reqwest(err) => { err.is_timeout() || err.is_connect() || err.status().map_or(false, |status| status.is_server_error()) } _ => false, } } } impl Default for Remote { fn default() -> Self { let (req_send, req_recv) = std::sync::mpsc::sync_channel(0); let (res_send, res_recv) = std::sync::mpsc::sync_channel(0); let handle = std::thread::spawn(move || -> Result<(), Error> { // We may error while configuring, which is expected as part of the internal protocol. The error will be // received and the sender of the request might restart us. let client = reqwest::blocking::ClientBuilder::new() .connect_timeout(std::time::Duration::from_secs(20)) .http1_title_case_headers() .build()?; for Request { url, headers, upload_body_kind, config, } in req_recv { let mut req_builder = if upload_body_kind.is_some() { client.post(url) } else { client.get(url) } .headers(headers); let (post_body_tx, mut post_body_rx) = pipe::unidirectional(0); let (mut response_body_tx, response_body_rx) = pipe::unidirectional(0); let (mut headers_tx, headers_rx) = pipe::unidirectional(0); if res_send .send(Response { headers: headers_rx, body: response_body_rx, upload_body: post_body_tx, }) .is_err() { // This means our internal protocol is violated as the one who sent the request isn't listening anymore. // Shut down as something is off. break; } req_builder = match upload_body_kind { Some(PostBodyDataKind::BoundedAndFitsIntoMemory) => { let mut buf = Vec::::with_capacity(512); post_body_rx.read_to_end(&mut buf)?; req_builder.body(buf) } Some(PostBodyDataKind::Unbounded) => req_builder.body(reqwest::blocking::Body::new(post_body_rx)), None => req_builder, }; let mut req = req_builder.build()?; if let Some(ref mut request_options) = config.backend.as_ref().and_then(|backend| backend.lock().ok()) { if let Some(options) = request_options.downcast_mut::() { if let Some(configure_request) = &mut options.configure_request { configure_request(&mut req)?; } } } let mut res = match client.execute(req).and_then(|res| res.error_for_status()) { Ok(res) => res, Err(err) => { let (kind, err) = match err.status() { Some(status) => { let kind = if status == reqwest::StatusCode::UNAUTHORIZED { std::io::ErrorKind::PermissionDenied } else if status.is_server_error() { std::io::ErrorKind::ConnectionAborted } else { std::io::ErrorKind::Other }; (kind, format!("Received HTTP status {}", status.as_str())) } None => (std::io::ErrorKind::Other, err.to_string()), }; let err = Err(std::io::Error::new(kind, err)); headers_tx.channel.send(err).ok(); continue; } }; let send_headers = { let headers = res.headers(); move || -> std::io::Result<()> { for (name, value) in headers { headers_tx.write_all(name.as_str().as_bytes())?; headers_tx.write_all(b":")?; headers_tx.write_all(value.as_bytes())?; headers_tx.write_all(b"\n")?; } // Make sure this is an FnOnce closure to signal the remote reader we are done. drop(headers_tx); Ok(()) } }; // We don't have to care if anybody is receiving the header, as a matter of fact we cannot fail sending them. // Thus an error means the receiver failed somehow, but might also have decided not to read headers at all. Fine with us. send_headers().ok(); // reading the response body is streaming and may fail for many reasons. If so, we send the error over the response // body channel and that's all we can do. if let Err(err) = std::io::copy(&mut res, &mut response_body_tx) { response_body_tx.channel.send(Err(err)).ok(); } } Ok(()) }); Remote { handle: Some(handle), request: req_send, response: res_recv, config: http::Options::default(), } } } /// utilities impl Remote { fn make_request( &mut self, url: &str, _base_url: &str, headers: impl IntoIterator>, upload_body_kind: Option, ) -> Result, http::Error> { let mut header_map = reqwest::header::HeaderMap::new(); for header_line in headers { let header_line = header_line.as_ref(); let colon_pos = header_line .find(':') .expect("header line must contain a colon to separate key and value"); let header_name = &header_line[..colon_pos]; let value = &header_line[colon_pos + 1..]; match reqwest::header::HeaderName::from_str(header_name) .ok() .zip(reqwest::header::HeaderValue::try_from(value.trim()).ok()) { Some((key, val)) => header_map.insert(key, val), None => continue, }; } self.request .send(Request { url: url.to_owned(), headers: header_map, upload_body_kind, config: self.config.clone(), }) .expect("the remote cannot be down at this point"); let Response { headers, body, upload_body, } = match self.response.recv() { Ok(res) => res, Err(_) => { let err = self .handle .take() .expect("always present") .join() .expect("no panic") .expect_err("no receiver means thread is down with init error"); *self = Self::default(); return Err(http::Error::InitHttpClient { source: Box::new(err) }); } }; Ok(http::PostResponse { post_body: upload_body, headers, body, }) } } impl http::Http for Remote { type Headers = pipe::Reader; type ResponseBody = pipe::Reader; type PostBody = pipe::Writer; fn get( &mut self, url: &str, base_url: &str, headers: impl IntoIterator>, ) -> Result, http::Error> { self.make_request(url, base_url, headers, None).map(Into::into) } fn post( &mut self, url: &str, base_url: &str, headers: impl IntoIterator>, post_body_kind: PostBodyDataKind, ) -> Result, http::Error> { self.make_request(url, base_url, headers, Some(post_body_kind)) } fn configure(&mut self, config: &dyn Any) -> Result<(), Box> { if let Some(config) = config.downcast_ref::() { self.config = config.clone(); } Ok(()) } } pub(crate) struct Request { pub url: String, pub headers: reqwest::header::HeaderMap, pub upload_body_kind: Option, pub config: http::Options, } /// A link to a thread who provides data for the contained readers. /// The expected order is: /// - write `upload_body` /// - read `headers` to end /// - read `body` to hend pub(crate) struct Response { pub headers: pipe::Reader, pub body: pipe::Reader, pub upload_body: pipe::Writer, }