707 lines
27 KiB
Rust
707 lines
27 KiB
Rust
use std::future::{Future, poll_fn};
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
|
use std::time::Duration;
|
|
|
|
use dever_runtime::{bytes::Bytes, http, net, sse, task, tls};
|
|
use rustls::RootCertStore;
|
|
use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject};
|
|
use tokio::io::AsyncReadExt;
|
|
|
|
const ROOT: &[u8] = include_bytes!("../fixtures/tls/root.pem");
|
|
const CERT: &[u8] = include_bytes!("../fixtures/tls/server.pem");
|
|
const KEY: &[u8] = include_bytes!("../fixtures/tls/server-key.pem");
|
|
|
|
fn bytes(value: &[u8]) -> Bytes {
|
|
Bytes::new(value.to_vec())
|
|
}
|
|
|
|
fn limits(timeout_ms: i64) -> http::Limits {
|
|
http::Limits {
|
|
header_bytes: 8192,
|
|
body_bytes: 65536,
|
|
timeout_ms,
|
|
connections: 2,
|
|
http2: Some(http::Http2Limits {
|
|
streams: 4,
|
|
stream_window_bytes: 65535,
|
|
connection_window_bytes: 65535,
|
|
}),
|
|
}
|
|
}
|
|
|
|
fn live() -> http::LiveLimits {
|
|
http::LiveLimits {
|
|
chunk_bytes: 1024,
|
|
idle_ms: 1000,
|
|
heartbeat_ms: 30,
|
|
}
|
|
}
|
|
|
|
fn run(future: impl Future<Output = Result<(), String>>) {
|
|
task::run_entry_with(
|
|
task::RuntimeConfig {
|
|
worker_threads: 1,
|
|
max_blocking_threads: 2,
|
|
task_capacity: 64,
|
|
},
|
|
async {
|
|
tokio::time::timeout(Duration::from_secs(8), future)
|
|
.await
|
|
.map_err(|_| "HTTP/2 server test deadline exceeded".to_owned())?
|
|
},
|
|
)
|
|
.unwrap();
|
|
}
|
|
|
|
fn error(error: impl std::fmt::Display) -> String {
|
|
error.to_string()
|
|
}
|
|
|
|
fn request(port: i64, target: &str) -> hyper::Request<()> {
|
|
hyper::Request::builder()
|
|
.version(hyper::Version::HTTP_2)
|
|
.uri(format!("http://localhost:{port}{target}"))
|
|
.body(())
|
|
.unwrap()
|
|
}
|
|
|
|
async fn connect(
|
|
port: i64,
|
|
) -> Result<(h2::client::SendRequest<bytes::Bytes>, task::Task<()>), String> {
|
|
let socket = tokio::net::TcpStream::connect(("127.0.0.1", port as u16))
|
|
.await
|
|
.map_err(error)?;
|
|
let (sender, connection) = h2::client::handshake(socket).await.map_err(error)?;
|
|
let driver = task::run(async move { connection.await.map_err(error) }).await?;
|
|
Ok((sender, driver))
|
|
}
|
|
|
|
async fn connect_with_window(
|
|
port: i64,
|
|
stream_window: u32,
|
|
) -> Result<(h2::client::SendRequest<bytes::Bytes>, task::Task<()>), String> {
|
|
let socket = tokio::net::TcpStream::connect(("127.0.0.1", port as u16))
|
|
.await
|
|
.map_err(error)?;
|
|
let mut builder = h2::client::Builder::new();
|
|
builder.initial_window_size(stream_window);
|
|
let (sender, connection) = builder
|
|
.handshake::<_, bytes::Bytes>(socket)
|
|
.await
|
|
.map_err(error)?;
|
|
let driver = task::run(async move { connection.await.map_err(error) }).await?;
|
|
Ok((sender, driver))
|
|
}
|
|
|
|
async fn start_request(
|
|
sender: h2::client::SendRequest<bytes::Bytes>,
|
|
request: hyper::Request<()>,
|
|
) -> Result<
|
|
(
|
|
h2::client::SendRequest<bytes::Bytes>,
|
|
h2::client::ResponseFuture,
|
|
),
|
|
String,
|
|
> {
|
|
let mut sender = sender.ready().await.map_err(error)?;
|
|
let (response, _) = sender.send_request(request, true).map_err(error)?;
|
|
Ok((sender, response))
|
|
}
|
|
|
|
async fn response_body(response: hyper::Response<h2::RecvStream>) -> Result<Vec<u8>, String> {
|
|
let mut body = response.into_body();
|
|
let mut result = Vec::new();
|
|
while let Some(chunk) = body.data().await {
|
|
result.extend_from_slice(&chunk.map_err(error)?);
|
|
}
|
|
Ok(result)
|
|
}
|
|
|
|
async fn wait_until(mut condition: impl FnMut() -> bool) -> Result<(), String> {
|
|
tokio::time::timeout(Duration::from_secs(1), async {
|
|
while !condition() {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.map_err(|_| "condition was not reached".to_owned())
|
|
}
|
|
|
|
#[test]
|
|
fn buffered_server_multiplexes_and_preserves_http2_request_semantics() {
|
|
run(async {
|
|
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
|
let invoked = Arc::new(AtomicUsize::new(0));
|
|
let route_gate = gate.clone();
|
|
let route_invoked = invoked.clone();
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve(
|
|
move |request| {
|
|
let gate = route_gate.clone();
|
|
let invoked = route_invoked.clone();
|
|
dever_runtime::api::scoped_request(request, move |request| async move {
|
|
invoked.fetch_add(1, Ordering::SeqCst);
|
|
let path = request.target.split('?').next().unwrap().to_owned();
|
|
let request_id = dever_runtime::api::request_id()?;
|
|
assert_eq!(
|
|
dever_runtime::api::client_address()?,
|
|
Some("127.0.0.1".into())
|
|
);
|
|
assert!(request.headers.iter().any(|header| {
|
|
header.name == "host"
|
|
&& header.value.values() == format!("localhost:{port}").as_bytes()
|
|
}));
|
|
if request.target.starts_with("/slow") {
|
|
let _ = gate.acquire().await.map_err(error)?;
|
|
}
|
|
assert_eq!(dever_runtime::api::path()?, path);
|
|
assert_eq!(dever_runtime::api::request_id()?, request_id);
|
|
dever_runtime::api::response_header("x-api-path", &path)?;
|
|
let response = http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: Bytes::from_string(request.target),
|
|
};
|
|
dever_runtime::api::commit_response_metadata()?;
|
|
Ok(response)
|
|
})
|
|
},
|
|
owner,
|
|
http::Limits {
|
|
body_bytes: 16,
|
|
..limits(500)
|
|
},
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, slow) = start_request(sender, request(port, "/slow?value=1")).await?;
|
|
let (sender, fast) = start_request(sender, request(port, "/fast")).await?;
|
|
let fast = tokio::time::timeout(Duration::from_millis(200), fast)
|
|
.await
|
|
.map_err(|_| "HTTP/2 streams were serialized".to_owned())?
|
|
.map_err(error)?;
|
|
assert_eq!(fast.status(), hyper::StatusCode::OK);
|
|
assert_eq!(fast.headers().get("x-api-path").unwrap(), "/fast");
|
|
assert_eq!(response_body(fast).await?, b"/fast");
|
|
gate.add_permits(1);
|
|
let slow = slow.await.map_err(error)?;
|
|
assert_eq!(slow.status(), hyper::StatusCode::OK);
|
|
assert_eq!(slow.headers().get("x-api-path").unwrap(), "/slow");
|
|
assert_eq!(response_body(slow).await?, b"/slow?value=1");
|
|
|
|
let mismatched = hyper::Request::builder()
|
|
.version(hyper::Version::HTTP_2)
|
|
.uri(format!("http://localhost:{port}/mismatch"))
|
|
.header(hyper::header::HOST, "other.example")
|
|
.body(())
|
|
.unwrap();
|
|
let (sender, rejected) = start_request(sender, mismatched).await?;
|
|
let rejected = rejected.await.map_err(error)?;
|
|
assert_eq!(rejected.status(), hyper::StatusCode::BAD_REQUEST);
|
|
assert!(!rejected.headers().contains_key(hyper::header::CONNECTION));
|
|
assert_eq!(response_body(rejected).await?, b"");
|
|
assert_eq!(invoked.load(Ordering::SeqCst), 2);
|
|
|
|
let mut sender = sender.ready().await.map_err(error)?;
|
|
let oversized = hyper::Request::builder()
|
|
.method(hyper::Method::POST)
|
|
.version(hyper::Version::HTTP_2)
|
|
.uri(format!("http://localhost:{port}/oversized"))
|
|
.body(())
|
|
.unwrap();
|
|
let (oversized, mut upload) = sender.send_request(oversized, false).map_err(error)?;
|
|
upload
|
|
.send_data(bytes::Bytes::from_static(b"12345678901234567"), true)
|
|
.map_err(error)?;
|
|
drop(upload);
|
|
let oversized = oversized.await.map_err(error)?;
|
|
assert_eq!(oversized.status(), hyper::StatusCode::PAYLOAD_TOO_LARGE);
|
|
assert_eq!(response_body(oversized).await?, b"");
|
|
assert_eq!(invoked.load(Ordering::SeqCst), 2);
|
|
|
|
let (sender, neighbor) = start_request(sender, request(port, "/neighbor")).await?;
|
|
assert_eq!(
|
|
response_body(neighbor.await.map_err(error)?).await?,
|
|
b"/neighbor"
|
|
);
|
|
assert_eq!(invoked.load(Ordering::SeqCst), 3);
|
|
drop(sender);
|
|
task::wait(driver).await?;
|
|
listener.close()?;
|
|
task::wait(server).await
|
|
});
|
|
}
|
|
|
|
struct Finished(Arc<AtomicUsize>);
|
|
|
|
impl Drop for Finished {
|
|
fn drop(&mut self) {
|
|
self.0.fetch_add(1, Ordering::SeqCst);
|
|
}
|
|
}
|
|
|
|
struct Marked(Arc<AtomicBool>);
|
|
|
|
impl Drop for Marked {
|
|
fn drop(&mut self) {
|
|
self.0.store(true, Ordering::SeqCst);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn resetting_one_live_stream_keeps_sse_neighbor_and_drains_blocking_work() {
|
|
run(async {
|
|
let finished = Arc::new(AtomicUsize::new(0));
|
|
let blocking_started = Arc::new(AtomicBool::new(false));
|
|
let blocking_done = Arc::new(AtomicBool::new(false));
|
|
let route_finished = finished.clone();
|
|
let route_blocking_started = blocking_started.clone();
|
|
let route_blocking_done = blocking_done.clone();
|
|
let stop_gate = Arc::new(tokio::sync::Semaphore::new(0));
|
|
let stop_started = Arc::new(AtomicBool::new(false));
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let owner_stop_gate = stop_gate.clone();
|
|
let owner_stop_started = stop_started.clone();
|
|
let server_owner = task::run(async move {
|
|
let server = task::run(http::serve_live(
|
|
move |request, reply| {
|
|
let finished = route_finished.clone();
|
|
let blocking_started = route_blocking_started.clone();
|
|
let blocking_done = route_blocking_done.clone();
|
|
async move {
|
|
let _finished = Finished(finished.clone());
|
|
match request.target.as_str() {
|
|
"/pending" => {
|
|
task::parallel(move || {
|
|
blocking_started.store(true, Ordering::SeqCst);
|
|
std::thread::sleep(Duration::from_millis(500));
|
|
blocking_done.store(true, Ordering::SeqCst);
|
|
Ok(())
|
|
})
|
|
.await?;
|
|
std::future::pending().await
|
|
}
|
|
"/events" => {
|
|
sse::start(&reply, vec![]).await?;
|
|
sse::send(
|
|
&reply,
|
|
sse::Event {
|
|
event: "update".into(),
|
|
data: "neighbor".into(),
|
|
id: None,
|
|
retry_ms: None,
|
|
},
|
|
)
|
|
.await?;
|
|
std::future::pending().await
|
|
}
|
|
_ => {
|
|
http::respond(
|
|
&reply,
|
|
http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"still-open"),
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
}
|
|
}
|
|
},
|
|
owner,
|
|
limits(1000),
|
|
live(),
|
|
))
|
|
.await?;
|
|
let _ = owner_stop_gate.acquire().await.map_err(error)?;
|
|
owner_stop_started.store(true, Ordering::SeqCst);
|
|
task::stop(server).await
|
|
})
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, pending) = start_request(sender, request(port, "/pending")).await?;
|
|
wait_until(|| blocking_started.load(Ordering::SeqCst)).await?;
|
|
drop(pending);
|
|
|
|
let (sender, events) = start_request(sender, request(port, "/events")).await?;
|
|
let mut events = events.await.map_err(error)?.into_body();
|
|
assert_eq!(
|
|
events.data().await.unwrap().map_err(error)?,
|
|
b"event: update\ndata: neighbor\n\n"[..]
|
|
);
|
|
|
|
let (sender, final_response) = start_request(sender, request(port, "/after-reset")).await?;
|
|
assert_eq!(
|
|
response_body(final_response.await.map_err(error)?).await?,
|
|
b"still-open"
|
|
);
|
|
drop(events);
|
|
stop_gate.add_permits(1);
|
|
wait_until(|| stop_started.load(Ordering::SeqCst)).await?;
|
|
task::stop(server_owner).await?;
|
|
assert!(blocking_done.load(Ordering::SeqCst));
|
|
drop(sender);
|
|
let _ = task::wait(driver).await;
|
|
listener.close()?;
|
|
assert!(finished.load(Ordering::SeqCst) >= 3);
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn request_faults_return_500_without_stopping_the_http2_connection() {
|
|
run(async {
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve(
|
|
|request| async move {
|
|
if request.target == "/fault" {
|
|
return Err("buffered handler fault".to_owned());
|
|
}
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"still-open"),
|
|
})
|
|
},
|
|
owner,
|
|
limits(500),
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, response) = start_request(sender, request(port, "/fault")).await?;
|
|
let response = response.await.map_err(error)?;
|
|
assert_eq!(response.status(), hyper::StatusCode::INTERNAL_SERVER_ERROR);
|
|
let (sender, response) = start_request(sender, request(port, "/after-fault")).await?;
|
|
assert_eq!(
|
|
response_body(response.await.map_err(error)?).await?,
|
|
b"still-open"
|
|
);
|
|
drop(sender);
|
|
task::stop(server).await?;
|
|
let _ = task::wait(driver).await;
|
|
listener.close()?;
|
|
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve_live(
|
|
|request, reply| async move {
|
|
if request.target == "/fault" {
|
|
return Err("live handler fault".to_owned());
|
|
}
|
|
http::respond(
|
|
&reply,
|
|
http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"still-open"),
|
|
},
|
|
)
|
|
.await
|
|
},
|
|
owner,
|
|
limits(500),
|
|
live(),
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, response) = start_request(sender, request(port, "/fault")).await?;
|
|
let response = response.await.map_err(error)?;
|
|
assert_eq!(response.status(), hyper::StatusCode::INTERNAL_SERVER_ERROR);
|
|
let (sender, response) = start_request(sender, request(port, "/after-fault")).await?;
|
|
assert_eq!(
|
|
response_body(response.await.map_err(error)?).await?,
|
|
b"still-open"
|
|
);
|
|
drop(sender);
|
|
task::stop(server).await?;
|
|
let _ = task::wait(driver).await;
|
|
listener.close()?;
|
|
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve(
|
|
|request| async move {
|
|
if request.target != "/invalid-response" {
|
|
return Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"still-open"),
|
|
});
|
|
}
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![http::Header {
|
|
name: "te".into(),
|
|
value: bytes(b"trailers"),
|
|
}],
|
|
body: bytes(b""),
|
|
})
|
|
},
|
|
owner,
|
|
limits(500),
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, response) = start_request(sender, request(port, "/invalid-response")).await?;
|
|
let response = response.await.map_err(error)?;
|
|
assert_eq!(response.status(), hyper::StatusCode::INTERNAL_SERVER_ERROR);
|
|
let (sender, response) = start_request(sender, request(port, "/after-invalid")).await?;
|
|
assert_eq!(
|
|
response_body(response.await.map_err(error)?).await?,
|
|
b"still-open"
|
|
);
|
|
drop(sender);
|
|
task::stop(server).await?;
|
|
let _ = task::wait(driver).await;
|
|
listener.close()?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn flow_controlled_stream_does_not_block_its_neighbor_and_resumes_on_window_update() {
|
|
run(async {
|
|
let body_finished = Arc::new(AtomicBool::new(false));
|
|
let reset_dropped = Arc::new(AtomicBool::new(false));
|
|
let reset_writes = Arc::new(AtomicUsize::new(0));
|
|
let route_finished = body_finished.clone();
|
|
let route_reset_dropped = reset_dropped.clone();
|
|
let route_reset_writes = reset_writes.clone();
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve_live(
|
|
move |request, reply| {
|
|
let body_finished = route_finished.clone();
|
|
let reset_dropped = route_reset_dropped.clone();
|
|
let reset_writes = route_reset_writes.clone();
|
|
async move {
|
|
match request.target.as_str() {
|
|
"/large" => {
|
|
http::start(&reply, 200, vec![]).await?;
|
|
for _ in 0..4 {
|
|
http::write(&reply, Bytes::new(vec![7; 16384])).await?;
|
|
}
|
|
http::finish(&reply).await?;
|
|
body_finished.store(true, Ordering::SeqCst);
|
|
Ok(())
|
|
}
|
|
"/reset-write" => {
|
|
let _dropped = Marked(reset_dropped);
|
|
http::start(&reply, 200, vec![]).await?;
|
|
for _ in 0..16 {
|
|
reset_writes.fetch_add(1, Ordering::SeqCst);
|
|
http::write(&reply, Bytes::new(vec![8; 16384])).await?;
|
|
}
|
|
http::finish(&reply).await
|
|
}
|
|
_ => {
|
|
http::respond(
|
|
&reply,
|
|
http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"neighbor"),
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
}
|
|
}
|
|
},
|
|
owner,
|
|
limits(500),
|
|
http::LiveLimits {
|
|
chunk_bytes: 16384,
|
|
..live()
|
|
},
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect_with_window(port, 1024).await?;
|
|
let (sender, large) = start_request(sender, request(port, "/large")).await?;
|
|
let mut large = large.await.map_err(error)?.into_body();
|
|
let (sender, neighbor) = start_request(sender, request(port, "/neighbor")).await?;
|
|
let neighbor = tokio::time::timeout(Duration::from_millis(200), neighbor)
|
|
.await
|
|
.map_err(|_| "flow-controlled stream blocked its neighbor".to_owned())?
|
|
.map_err(error)?;
|
|
assert_eq!(response_body(neighbor).await?, b"neighbor");
|
|
assert!(!body_finished.load(Ordering::SeqCst));
|
|
|
|
let mut received = 0;
|
|
let mut flow = large.flow_control().clone();
|
|
while let Some(chunk) = large.data().await {
|
|
let chunk = chunk.map_err(error)?;
|
|
received += chunk.len();
|
|
flow.release_capacity(chunk.len()).map_err(error)?;
|
|
}
|
|
assert_eq!(received, 4 * 16384);
|
|
wait_until(|| body_finished.load(Ordering::SeqCst)).await?;
|
|
drop(flow);
|
|
drop(large);
|
|
|
|
let (sender, reset) = start_request(sender, request(port, "/reset-write")).await?;
|
|
let reset = reset.await.map_err(error)?.into_body();
|
|
wait_until(|| reset_writes.load(Ordering::SeqCst) >= 3).await?;
|
|
tokio::time::sleep(Duration::from_millis(30)).await;
|
|
assert!(reset_writes.load(Ordering::SeqCst) < 16);
|
|
drop(reset);
|
|
wait_until(|| reset_dropped.load(Ordering::SeqCst)).await?;
|
|
let (sender, neighbor) = start_request(sender, request(port, "/after-reset-write")).await?;
|
|
assert_eq!(
|
|
response_body(neighbor.await.map_err(error)?).await?,
|
|
b"neighbor"
|
|
);
|
|
|
|
drop(sender);
|
|
task::wait(driver).await?;
|
|
listener.close()?;
|
|
task::wait(server).await
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn listener_close_sends_goaway_and_drains_an_active_stream() {
|
|
run(async {
|
|
let entered = Arc::new(AtomicBool::new(false));
|
|
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
|
let route_entered = entered.clone();
|
|
let route_gate = gate.clone();
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve(
|
|
move |_| {
|
|
let entered = route_entered.clone();
|
|
let gate = route_gate.clone();
|
|
async move {
|
|
entered.store(true, Ordering::SeqCst);
|
|
let _ = gate.acquire().await.map_err(error)?;
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"drained"),
|
|
})
|
|
}
|
|
},
|
|
owner,
|
|
limits(500),
|
|
))
|
|
.await?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (mut sender, response) = start_request(sender, request(port, "/active")).await?;
|
|
wait_until(|| entered.load(Ordering::SeqCst)).await?;
|
|
listener.close()?;
|
|
tokio::time::timeout(Duration::from_millis(500), async {
|
|
loop {
|
|
if poll_fn(|context| sender.poll_ready(context)).await.is_err() {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.map_err(|_| "client did not observe server GOAWAY".to_owned())?;
|
|
gate.add_permits(1);
|
|
assert_eq!(
|
|
response_body(response.await.map_err(error)?).await?,
|
|
b"drained"
|
|
);
|
|
drop(sender);
|
|
task::wait(driver).await?;
|
|
task::wait(server).await
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn silent_peer_times_out_and_tls_negotiates_h2() {
|
|
run(async {
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve(
|
|
|_| async {
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"ok"),
|
|
})
|
|
},
|
|
owner,
|
|
limits(80),
|
|
))
|
|
.await?;
|
|
let mut silent = tokio::net::TcpStream::connect(("127.0.0.1", port as u16))
|
|
.await
|
|
.map_err(error)?;
|
|
tokio::time::sleep(Duration::from_millis(160)).await;
|
|
let mut received = Vec::new();
|
|
tokio::time::timeout(
|
|
Duration::from_millis(500),
|
|
silent.read_to_end(&mut received),
|
|
)
|
|
.await
|
|
.map_err(|_| "silent HTTP/2 peer was not closed after its deadline".to_owned())?
|
|
.map_err(error)?;
|
|
let (sender, driver) = connect(port).await?;
|
|
let (sender, response) = start_request(sender, request(port, "/after-timeout")).await?;
|
|
assert_eq!(response_body(response.await.map_err(error)?).await?, b"ok");
|
|
drop(sender);
|
|
task::wait(driver).await?;
|
|
listener.close()?;
|
|
task::wait(server).await?;
|
|
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let owner = listener.clone();
|
|
let server = task::run(http::serve_tls(
|
|
|_| async {
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"tls"),
|
|
})
|
|
},
|
|
owner,
|
|
limits(500),
|
|
tls::server(&bytes(CERT), &bytes(KEY))?,
|
|
))
|
|
.await?;
|
|
let socket = tokio::net::TcpStream::connect(("127.0.0.1", port as u16))
|
|
.await
|
|
.map_err(error)?;
|
|
let mut roots = RootCertStore::empty();
|
|
for certificate in CertificateDer::pem_slice_iter(ROOT) {
|
|
roots.add(certificate.map_err(error)?).map_err(error)?;
|
|
}
|
|
let mut config = rustls::ClientConfig::builder()
|
|
.with_root_certificates(roots)
|
|
.with_no_client_auth();
|
|
config.alpn_protocols = vec![b"h2".to_vec()];
|
|
let name = ServerName::try_from("localhost".to_owned()).map_err(error)?;
|
|
let stream = tokio_rustls::TlsConnector::from(Arc::new(config))
|
|
.connect(name, socket)
|
|
.await
|
|
.map_err(error)?;
|
|
assert_eq!(stream.get_ref().1.alpn_protocol(), Some(b"h2".as_slice()));
|
|
let (sender, connection) = h2::client::handshake(stream).await.map_err(error)?;
|
|
let driver = task::run(async move { connection.await.map_err(error) }).await?;
|
|
let (sender, response) = start_request(sender, request(port, "/tls")).await?;
|
|
assert_eq!(response_body(response.await.map_err(error)?).await?, b"tls");
|
|
drop(sender);
|
|
task::wait(driver).await?;
|
|
listener.close()?;
|
|
task::wait(server).await
|
|
});
|
|
}
|