use crate::stats::{RealtimeStats, Statistics};
use crate::{Options, fatal};
use bytes::Bytes;
use http::{Request, header};
use http::header::HeaderValue;
use http_body_util::{BodyExt, Either, Full};
use hyper::StatusCode;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Instant;
use crate::client::utils::{
build_headers, build_http_connection_legacy, build_trailers, discard_body, should_stop,
};
pub async fn http_hyper_legacy(
tid: usize,
cid: usize,
opts: Arc<Options>,
rt_stats: &RealtimeStats,
) -> Statistics {
let mut statistics = Statistics::new(opts.latency);
let mut total: u32 = 0;
let mut banner = HashSet::new();
let uri_str = opts.uri[cid % opts.uri.len()].as_str();
let uri = uri_str
.parse::<http::Uri>()
.unwrap_or_else(|e| fatal!(1, "invalid uri: {e}"));
let headers = build_headers(&uri, opts.as_ref())
.unwrap_or_else(|e| fatal!(2, "could not build headers: {e}"));
let trailers = build_trailers(opts.as_ref())
.unwrap_or_else(|e| fatal!(2, "could not build trailers: {e}"));
let bodies: Vec<Full<Bytes>> = opts.bodies().map_or_else(
|e| fatal!(2, "could not read body: {e}"),
|v| v.into_iter().map(Full::new).collect(),
);
let clock = quanta::Clock::new();
let start = Instant::now();
'connection: loop {
if should_stop(total, start, &opts) {
break 'connection;
}
if cid < opts.uri.len() && !banner.contains(uri_str) {
banner.insert(uri_str.to_owned());
println!(
"hyper-legacy [{tid:>2}] -> connecting. {} {} {}...",
opts.method.as_ref().unwrap_or(&http::Method::GET),
uri,
if opts.http2 { "HTTP/2" } else { "HTTP/1.1" }
);
}
let mut client = build_http_connection_legacy(&opts);
statistics.inc_conn();
let mut conn_req_count: u32 = 0;
loop {
conn_req_count += 1;
let is_last = conn_req_count >= opts.rpc;
let body = bodies
.get(total as usize)
.or(bodies.last())
.cloned()
.unwrap_or_else(|| Full::new(Bytes::from("")));
let body = match &trailers {
None => Either::Left(body.clone()),
tr => {
let trailers = tr.clone().map(Ok);
Either::Right(body.clone().with_trailers(std::future::ready(trailers)))
}
};
let mut req = Request::new(body);
*req.method_mut() = opts.method.clone().unwrap_or(http::Method::GET);
*req.uri_mut() = uri.clone();
let mut req_headers = headers.clone();
if is_last {
req_headers.insert(header::CONNECTION, HeaderValue::from_static("close"));
}
*req.headers_mut() = req_headers;
let start_lat = opts.latency.then_some(clock.raw());
match client.request(req).await {
Ok(res) => match discard_body(res).await {
Ok(StatusCode::OK) => statistics.inc_ok(rt_stats),
Ok(code) => statistics.set_http_status(code, rt_stats),
Err(ref err) => {
statistics.set_error(err.as_ref(), rt_stats);
total += 1;
continue 'connection;
}
},
Err(ref err) => {
statistics.set_error(err, rt_stats);
total += 1;
continue 'connection;
}
}
if let Some(start_lat) = start_lat
&& let Some(hist) = &mut statistics.latency
{
hist.record(clock.delta_as_nanos(start_lat, clock.raw()) / 1000)
.ok();
};
total += 1;
if should_stop(total, start, &opts) {
break 'connection;
}
if is_last {
client = build_http_connection_legacy(&opts);
conn_req_count = 0;
}
}
}
statistics
}