Skip to content

feat(rust): support streaming request bodies #514

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
barjin wants to merge 2 commits into from
Open

feat(rust): support streaming request bodies #514

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cf61da1
feat(rust): support streaming request bodies
barjin Aug 13, 2026
35da2b2
chore: run formatter
barjin Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion impit-node/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
16 changes: 8 additions & 8 deletions impit-python/src/async_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())),
};

Expand Down
32 changes: 24 additions & 8 deletions impit-python/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())),
};

Expand Down
2 changes: 2 additions & 0 deletions impit/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
42 changes: 22 additions & 20 deletions impit/src/impit.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use crate::{
fingerprint::BrowserFingerprint,
http3::H3Engine,
http_headers::HttpHeaders,
request::{ImpitRequest, RequestOptions},
request::{ImpitBody, ImpitRequest, RequestOptions},
tls,
};

Expand All @@ -31,7 +31,7 @@ struct PreparedRequest {
method: Method,
url: Url,
headers: HeaderMap,
body: Option<Vec<u8>>,
body: ImpitBody,
}

impl<CookieStoreImpl: CookieStore + 'static> Default for Impit<CookieStoreImpl> {
Expand Down Expand Up @@ -401,7 +401,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
&self,
method: Method,
url: Url,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
headers: Vec<(String, String)>,
) -> ImpitRequest {
let host = url.host_str().unwrap_or_default().to_string();
Expand All @@ -416,7 +416,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {

ImpitRequest {
url,
body,
body: body.unwrap_or_default(),
headers: headers.iter().collect(),
method: method.to_string(),
}
Expand All @@ -425,7 +425,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
async fn execute_request(
&self,
client: &reqwest::Client,
prepared: &PreparedRequest,
prepared: &mut PreparedRequest,
timeout: Option<Duration>,
h3: bool,
) -> Result<Response, reqwest::Error> {
Expand All @@ -441,8 +441,8 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
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
Expand Down Expand Up @@ -486,14 +486,16 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
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,
Expand All @@ -512,7 +514,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
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);
};
Expand All @@ -521,7 +523,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
"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,
Expand Down Expand Up @@ -555,7 +557,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
&self,
method: Method,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
let url = self.parse_url(url)?;
Expand Down Expand Up @@ -585,7 +587,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn get(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::GET, url, body, options).await
Expand All @@ -600,7 +602,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn head(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::HEAD, url, body, options).await
Expand All @@ -615,7 +617,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn options(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::OPTIONS, url, body, options).await
Expand All @@ -630,7 +632,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn trace(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::TRACE, url, body, options).await
Expand All @@ -645,7 +647,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn delete(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::DELETE, url, body, options).await
Expand All @@ -660,7 +662,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn post(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::POST, url, body, options).await
Expand All @@ -675,7 +677,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn put(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::PUT, url, body, options).await
Expand All @@ -690,7 +692,7 @@ impl<CookieStoreImpl: CookieStore + 'static> Impit<CookieStoreImpl> {
pub async fn patch(
&self,
url: String,
body: Option<Vec<u8>>,
body: Option<ImpitBody>,
options: Option<RequestOptions>,
) -> Result<Response, ImpitError> {
self.make_request(Method::PATCH, url, body, options).await
Expand Down
60 changes: 59 additions & 1 deletion impit/src/request.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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<u8>),
/// 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<S>(stream: S) -> Self
where
S: TryStream + Send + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
Bytes: From<S::Ok>,
{
Self::Stream(reqwest::Body::wrap_stream(stream))
}

pub(crate) fn take(&mut self) -> Option<reqwest::Body> {
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<Vec<u8>> for ImpitBody {
fn from(bytes: Vec<u8>) -> Self {
Self::Bytes(bytes)
}
}

pub struct ImpitRequest {
pub url: Url,
pub body: Option<Vec<u8>>,
pub body: ImpitBody,
pub headers: Vec<(String, String)>,
pub method: String,
}