diff --git a/Cargo.lock b/Cargo.lock index 6cb3f9d6..2ae16d7b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1299,7 +1299,9 @@ dependencies = [ name = "impit" version = "0.1.0" dependencies = [ + "bytes", "encoding", + "futures-core", "hickory-proto", "hickory-resolver", "hyper", diff --git a/impit-node/src/lib.rs b/impit-node/src/lib.rs index 333c444a..90ec326d 100644 --- a/impit-node/src/lib.rs +++ b/impit-node/src/lib.rs @@ -138,7 +138,7 @@ impl ImpitWrapper { .unwrap_or_default(); let body = request_init .and_then(|init| init.body) - .map(|array| array.to_vec()); + .map(|array| array.to_vec().into()); let response = if matches!(method, HttpMethod::Get | HttpMethod::Head) && body.is_some() { Err(ImpitError::BindingPassthroughError( diff --git a/impit-python/src/async_client.rs b/impit-python/src/async_client.rs index 8963371a..24f7a6be 100644 --- a/impit-python/src/async_client.rs +++ b/impit-python/src/async_client.rs @@ -423,14 +423,14 @@ impl AsyncClient { pyo3_async_runtimes::tokio::future_into_py::<_, ImpitPyResponse>(py, async move { let response = match method_str.to_lowercase().as_str() { - "get" => impit.get(url, Some(body), Some(options)).await, - "post" => impit.post(url, Some(body), Some(options)).await, - "patch" => impit.patch(url, Some(body), Some(options)).await, - "put" => impit.put(url, Some(body), Some(options)).await, - "options" => impit.options(url, Some(body), Some(options)).await, - "trace" => impit.trace(url, Some(body), Some(options)).await, - "head" => impit.head(url, Some(body), Some(options)).await, - "delete" => impit.delete(url, Some(body), Some(options)).await, + "get" => impit.get(url, Some(body.into()), Some(options)).await, + "post" => impit.post(url, Some(body.into()), Some(options)).await, + "patch" => impit.patch(url, Some(body.into()), Some(options)).await, + "put" => impit.put(url, Some(body.into()), Some(options)).await, + "options" => impit.options(url, Some(body.into()), Some(options)).await, + "trace" => impit.trace(url, Some(body.into()), Some(options)).await, + "head" => impit.head(url, Some(body.into()), Some(options)).await, + "delete" => impit.delete(url, Some(body.into()), Some(options)).await, _ => Err(ImpitError::InvalidMethod(method_str.to_string())), }; diff --git a/impit-python/src/client.rs b/impit-python/src/client.rs index a817cbfe..06db785f 100644 --- a/impit-python/src/client.rs +++ b/impit-python/src/client.rs @@ -417,14 +417,30 @@ impl Client { py.detach(|| { pyo3_async_runtimes::tokio::get_runtime().block_on(async { let response = match method.to_lowercase().as_str() { - "get" => self.impit.get(url, Some(body), Some(options)).await, - "post" => self.impit.post(url, Some(body), Some(options)).await, - "patch" => self.impit.patch(url, Some(body), Some(options)).await, - "put" => self.impit.put(url, Some(body), Some(options)).await, - "options" => self.impit.options(url, Some(body), Some(options)).await, - "trace" => self.impit.trace(url, Some(body), Some(options)).await, - "head" => self.impit.head(url, Some(body), Some(options)).await, - "delete" => self.impit.delete(url, Some(body), Some(options)).await, + "get" => self.impit.get(url, Some(body.into()), Some(options)).await, + "post" => self.impit.post(url, Some(body.into()), Some(options)).await, + "patch" => { + self.impit + .patch(url, Some(body.into()), Some(options)) + .await + } + "put" => self.impit.put(url, Some(body.into()), Some(options)).await, + "options" => { + self.impit + .options(url, Some(body.into()), Some(options)) + .await + } + "trace" => { + self.impit + .trace(url, Some(body.into()), Some(options)) + .await + } + "head" => self.impit.head(url, Some(body.into()), Some(options)).await, + "delete" => { + self.impit + .delete(url, Some(body.into()), Some(options)) + .await + } _ => Err(ImpitError::InvalidMethod(method.to_string())), }; diff --git a/impit/Cargo.toml b/impit/Cargo.toml index 3c842817..1f5a9cb4 100644 --- a/impit/Cargo.toml +++ b/impit/Cargo.toml @@ -6,7 +6,9 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +bytes = "1" encoding = "0.2.33" +futures-core = "0.3" hickory-proto = "0.26.1" hickory-resolver = "0.26.1" log = "0.4.22" diff --git a/impit/src/impit.rs b/impit/src/impit.rs index ba43d156..cd0b5cf1 100644 --- a/impit/src/impit.rs +++ b/impit/src/impit.rs @@ -10,7 +10,7 @@ use crate::{ fingerprint::BrowserFingerprint, http3::H3Engine, http_headers::HttpHeaders, - request::{ImpitRequest, RequestOptions}, + request::{ImpitBody, ImpitRequest, RequestOptions}, tls, }; @@ -31,7 +31,7 @@ struct PreparedRequest { method: Method, url: Url, headers: HeaderMap, - body: Option>, + body: ImpitBody, } impl Default for Impit { @@ -401,7 +401,7 @@ impl Impit { &self, method: Method, url: Url, - body: Option>, + body: Option, headers: Vec<(String, String)>, ) -> ImpitRequest { let host = url.host_str().unwrap_or_default().to_string(); @@ -416,7 +416,7 @@ impl Impit { ImpitRequest { url, - body, + body: body.unwrap_or_default(), headers: headers.iter().collect(), method: method.to_string(), } @@ -425,7 +425,7 @@ impl Impit { async fn execute_request( &self, client: &reqwest::Client, - prepared: &PreparedRequest, + prepared: &mut PreparedRequest, timeout: Option, h3: bool, ) -> Result { @@ -441,8 +441,8 @@ impl Impit { req = req.timeout(t); } - if let Some(b) = prepared.body.clone() { - req = req.body(b); + if let Some(body) = prepared.body.take() { + req = req.body(body); } req.send().await @@ -486,14 +486,16 @@ impl Impit { RedirectBehavior::ManualRedirect => 0, }; - let prepared = PreparedRequest { + let mut prepared = PreparedRequest { method: method.clone(), url: request.url.clone(), headers: header_map, body: request.body, }; - let primary_result = self.execute_request(client, &prepared, timeout, h3).await; + let primary_result = self + .execute_request(client, &mut prepared, timeout, h3) + .await; let response = match primary_result { Ok(resp) => resp, @@ -512,7 +514,7 @@ impl Impit { let fallback_client = self .vanilla_client .as_ref() - .filter(|_| primary_error.is_connect_error()); + .filter(|_| primary_error.is_connect_error() && prepared.body.is_replayable()); let Some(vanilla_client) = fallback_client else { return Err(primary_error); }; @@ -521,7 +523,7 @@ impl Impit { "Primary request to {url} failed with {primary_error}, retrying with vanilla client" ); match self - .execute_request(vanilla_client, &prepared, timeout, false) + .execute_request(vanilla_client, &mut prepared, timeout, false) .await { Ok(resp) => resp, @@ -555,7 +557,7 @@ impl Impit { &self, method: Method, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { let url = self.parse_url(url)?; @@ -585,7 +587,7 @@ impl Impit { pub async fn get( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::GET, url, body, options).await @@ -600,7 +602,7 @@ impl Impit { pub async fn head( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::HEAD, url, body, options).await @@ -615,7 +617,7 @@ impl Impit { pub async fn options( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::OPTIONS, url, body, options).await @@ -630,7 +632,7 @@ impl Impit { pub async fn trace( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::TRACE, url, body, options).await @@ -645,7 +647,7 @@ impl Impit { pub async fn delete( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::DELETE, url, body, options).await @@ -660,7 +662,7 @@ impl Impit { pub async fn post( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::POST, url, body, options).await @@ -675,7 +677,7 @@ impl Impit { pub async fn put( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::PUT, url, body, options).await @@ -690,7 +692,7 @@ impl Impit { pub async fn patch( &self, url: String, - body: Option>, + body: Option, options: Option, ) -> Result { self.make_request(Method::PATCH, url, body, options).await diff --git a/impit/src/request.rs b/impit/src/request.rs index a0f1d0ad..e61b0c40 100644 --- a/impit/src/request.rs +++ b/impit/src/request.rs @@ -1,5 +1,7 @@ use std::time::Duration; +use bytes::Bytes; +use futures_core::TryStream; use url::Url; /// A struct that holds the request options. @@ -24,9 +26,65 @@ pub struct RequestOptions { pub http3_prior_knowledge: bool, } +/// The body of a request. +#[derive(Default)] +pub enum ImpitBody { + /// No request body. + #[default] + Empty, + /// A body that is fully buffered in memory before the request is sent. + Bytes(Vec), + /// A body that is streamed into the request as its chunks are produced. + /// + /// Note that streamed bodies can only be sent once, so requests using them are never retried. + Stream(reqwest::Body), + /// A streamed body that has already been sent and cannot be replayed. + Consumed, +} + +impl ImpitBody { + /// Creates a streaming body from a stream of byte chunks. + /// + /// Unlike [`ImpitBody::Bytes`], the chunks are sent as they are produced, so the whole body + /// never has to be held in memory. The request uses `Transfer-Encoding: chunked` unless a + /// `Content-Length` header is set explicitly. + pub fn from_stream(stream: S) -> Self + where + S: TryStream + Send + 'static, + S::Error: Into>, + Bytes: From, + { + Self::Stream(reqwest::Body::wrap_stream(stream)) + } + + pub(crate) fn take(&mut self) -> Option { + match std::mem::replace(self, Self::Consumed) { + Self::Bytes(bytes) => { + *self = Self::Bytes(bytes.clone()); + Some(bytes.into()) + } + Self::Stream(body) => Some(body), + body => { + *self = body; + None + } + } + } + + pub(crate) fn is_replayable(&self) -> bool { + !matches!(self, Self::Consumed) + } +} + +impl From> for ImpitBody { + fn from(bytes: Vec) -> Self { + Self::Bytes(bytes) + } +} + pub struct ImpitRequest { pub url: Url, - pub body: Option>, + pub body: ImpitBody, pub headers: Vec<(String, String)>, pub method: String, }