use super::*;
pub(super) async fn run_h1_stream_pump(mut pump: H1StreamPump) {
let initial = std::mem::take(&mut pump.initial_body);
let excess = has_excess(pump.framing, initial.len());
let drained_clean = stream_body_into(pump.io.as_mut(), pump.framing, initial, &pump.tx).await;
if drained_clean && pump.reusable && !excess {
pump.pool
.return_h1(pump.key, H1Slot { io: pump.io }, pump.tls);
if pump.count_install {
pump.pool.note_h1_install();
}
}
drop(pump.permit);
}
pub(super) async fn stream_body_into(
stream: &mut dyn H1Io,
framing: BodyFraming,
initial: Vec<u8>,
tx: &mpsc::Sender<io::Result<Bytes>>,
) -> bool {
let result = match framing {
BodyFraming::None => Ok(true),
BodyFraming::Fixed(len) => stream_fixed_into(stream, initial, len, tx).await,
BodyFraming::Chunked => stream_chunked_into(stream, initial, tx).await,
BodyFraming::ToClose => stream_to_close_into(stream, initial, tx).await,
};
match result {
Ok(clean) => clean,
Err(e) => {
let _ = tx.send(Err(e)).await;
false
}
}
}
pub(super) async fn stream_fixed_into(
stream: &mut dyn H1Io,
initial: Vec<u8>,
len: u64,
tx: &mpsc::Sender<io::Result<Bytes>>,
) -> io::Result<bool> {
let mut remaining = len;
if !initial.is_empty() {
let take = (initial.len() as u64).min(remaining) as usize;
if take > 0 {
if tx
.send(Ok(Bytes::copy_from_slice(&initial[..take])))
.await
.is_err()
{
return Ok(false);
}
remaining -= take as u64;
}
}
let mut tmp = vec![0u8; 8192];
while remaining > 0 {
let want = remaining.min(tmp.len() as u64) as usize;
let n = stream.read(&mut tmp[..want]).await?;
if n == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"connection closed before HTTP/1.1 body completed",
));
}
if tx
.send(Ok(Bytes::copy_from_slice(&tmp[..n])))
.await
.is_err()
{
return Ok(false);
}
remaining -= n as u64;
}
Ok(true)
}
pub(super) async fn stream_to_close_into(
stream: &mut dyn H1Io,
initial: Vec<u8>,
tx: &mpsc::Sender<io::Result<Bytes>>,
) -> io::Result<bool> {
if !initial.is_empty() && tx.send(Ok(Bytes::from(initial))).await.is_err() {
return Ok(false);
}
let mut tmp = vec![0u8; 8192];
loop {
let n = stream.read(&mut tmp).await?;
if n == 0 {
return Ok(true);
}
if tx
.send(Ok(Bytes::copy_from_slice(&tmp[..n])))
.await
.is_err()
{
return Ok(false);
}
}
}
pub(super) async fn stream_chunked_into(
stream: &mut dyn H1Io,
mut buf: Vec<u8>,
tx: &mpsc::Sender<io::Result<Bytes>>,
) -> io::Result<bool> {
loop {
let line_end = read_until_crlf(stream, &mut buf)
.await
.map_err(h1err_to_io)?;
let size_line = String::from_utf8_lossy(&buf[..line_end]);
let size_token = size_line.split(';').next().unwrap_or("").trim();
let mut remaining = u64::from_str_radix(size_token, 16).map_err(|e| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("invalid chunk size: {e}"),
)
})?;
buf.drain(..line_end + 2);
if remaining == 0 {
read_chunk_trailers(stream, &mut buf)
.await
.map_err(h1err_to_io)?;
return Ok(true);
}
while remaining > 0 {
if buf.is_empty() {
read_more(stream, &mut buf).await.map_err(h1err_to_io)?;
}
let take = usize::try_from(remaining).map_or(buf.len(), |wanted| wanted.min(buf.len()));
let piece = Bytes::copy_from_slice(&buf[..take]);
buf.drain(..take);
remaining -= take as u64;
if tx.send(Ok(piece)).await.is_err() {
return Ok(false);
}
}
read_until_available(stream, &mut buf, 2)
.await
.map_err(h1err_to_io)?;
if &buf[..2] != b"\r\n" {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"chunk missing CRLF terminator",
));
}
buf.drain(..2);
}
}
#[expect(
clippy::too_many_arguments,
reason = "flat per-request wire fields across one internal call path"
)]
pub(super) async fn send_request_h1_streaming(
pool: &Arc<Pool>,
connector: &FingerprintConnector,
scheme: &str,
host: &str,
port: u16,
method: &str,
url: &url::Url,
headers: Vec<(String, String)>,
body: H1Body,
proxy: Option<&str>,
target: H1Target,
permit: OwnedSemaphorePermit,
key: PoolKey,
) -> Result<H1Response, H1PooledError> {
let started = Instant::now();
let replay = replay_body(&body);
let mut body = body;
if let Some((slot, tls)) = checkout_live_h1(pool, &key) {
let pooled_body = std::mem::replace(&mut body, H1Body::Empty);
let mut io = slot.io;
match exchange_head_on_stream(
io.as_mut(),
method,
url,
headers.clone(),
pooled_body,
target,
)
.await
{
Ok((head, reusable)) => {
tracing::Span::current().record("pool.hit", true);
let (tx, rx) = mpsc::channel(STREAM_CHANNEL_DEPTH);
tokio::spawn(run_h1_stream_pump(H1StreamPump {
io,
permit,
pool: pool.clone(),
key,
tls: tls.clone(),
framing: head.framing,
initial_body: head.initial_body,
reusable,
count_install: false,
tx,
}));
return Ok(H1Response {
status: head.status,
headers: head.headers,
body: H1ResponseBody::Streaming(BodyStream::new(rx)),
tls: tls_for_scheme(scheme, &tls),
timing: ResponseTiming::leg(started, None),
});
}
Err(e) => body = resend_after_failure(pool, &key, method, replay, e)?,
}
}
tracing::Span::current().record("pool.hit", false);
let connect_started = Instant::now();
let (io, tls): (Box<dyn H1Io>, TlsInfo) =
open_new(connector, scheme, host, port, proxy).await?;
let connect_ms = ResponseTiming::millis(connect_started);
let mut slot = H1Slot { io };
let (head, reusable) =
exchange_head_on_stream(slot.io.as_mut(), method, url, headers, body, target).await?;
let (tx, rx) = mpsc::channel(STREAM_CHANNEL_DEPTH);
tokio::spawn(run_h1_stream_pump(H1StreamPump {
io: slot.io,
permit,
pool: pool.clone(),
key,
tls: tls.clone(),
framing: head.framing,
initial_body: head.initial_body,
reusable,
count_install: true,
tx,
}));
Ok(H1Response {
status: head.status,
headers: head.headers,
body: H1ResponseBody::Streaming(BodyStream::new(rx)),
tls: tls_for_scheme(scheme, &tls),
timing: ResponseTiming::leg(started, Some(connect_ms)),
})
}