1 + | /*
|
2 + | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 + | * SPDX-License-Identifier: Apache-2.0
|
4 + | */
|
5 + |
|
6 + | //! H2 connection pool behavior tests.
|
7 + | //!
|
8 + | //! Tests the v2 pool's HTTP/2 path: multiplexing, GOAWAY handling,
|
9 + | //! connection poisoning, and stream limits.
|
10 + | //!
|
11 + | //! Uses plain TCP with a fake ALPN signal (`Connected::new().negotiated_h2()`)
|
12 + | //! so no TLS infrastructure is needed. The pool's Negotiate layer trusts
|
13 + | //! the `Connected` metadata to route to the H2 path.
|
14 + |
|
15 + | #![cfg(all(
|
16 + | feature = "wire-mock",
|
17 + | feature = "default-client",
|
18 + | feature = "test-util",
|
19 + | aws_sdk_unstable
|
20 + | ))]
|
21 + |
|
22 + | use aws_smithy_http_client::pool::{Client, SharedPool};
|
23 + | use aws_smithy_runtime_api::client::http::{
|
24 + | HttpClient, HttpConnector, HttpConnectorSettings, SharedHttpClient,
|
25 + | };
|
26 + | use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
|
27 + | use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
|
28 + | use aws_smithy_runtime_api::shared::IntoShared;
|
29 + | use bytes::Bytes;
|
30 + | use h2::server::SendResponse;
|
31 + | use h2::RecvStream;
|
32 + | use http_body_util::BodyExt;
|
33 + | use hyper_util::client::legacy::connect::Connected;
|
34 + | use std::future::Future;
|
35 + | use std::net::SocketAddr;
|
36 + | use std::pin::Pin;
|
37 + | use std::sync::atomic::{AtomicUsize, Ordering};
|
38 + | use std::sync::Arc;
|
39 + | use std::task::{Context, Poll};
|
40 + | use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
|
41 + | use tokio::net::{TcpListener, TcpStream};
|
42 + | use tower::Service;
|
43 + |
|
44 + | // ---------------------------------------------------------------------------
|
45 + | // H2MockServer — plain TCP server speaking H2 via the h2 crate directly
|
46 + | // ---------------------------------------------------------------------------
|
47 + |
|
48 + | /// Handler function type for H2 requests.
|
49 + | type H2Handler =
|
50 + | Arc<dyn Fn(http_1x::Request<RecvStream>, SendResponse<Bytes>) + Send + Sync + 'static>;
|
51 + |
|
52 + | struct H2MockServer {
|
53 + | addr: SocketAddr,
|
54 + | /// Total H2 connections accepted (each connection can multiplex many streams).
|
55 + | connections: Arc<AtomicUsize>,
|
56 + | /// Total streams (requests) handled across all connections.
|
57 + | streams: Arc<AtomicUsize>,
|
58 + | _shutdown: tokio::sync::oneshot::Sender<()>,
|
59 + | }
|
60 + |
|
61 + | impl H2MockServer {
|
62 + | /// Start an H2 server that responds 200 with the given body to every request.
|
63 + | async fn start(body: &'static str) -> Self {
|
64 + | Self::start_with_handler(Arc::new(move |_req, mut respond| {
|
65 + | let response = http_1x::Response::builder().status(200).body(()).unwrap();
|
66 + | let mut send_stream = respond.send_response(response, false).unwrap();
|
67 + | send_stream.send_data(Bytes::from(body), true).unwrap();
|
68 + | }))
|
69 + | .await
|
70 + | }
|
71 + |
|
72 + | /// Start with a custom handler for each stream.
|
73 + | async fn start_with_handler(handler: H2Handler) -> Self {
|
74 + | let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
75 + | let addr = listener.local_addr().unwrap();
|
76 + | let connections = Arc::new(AtomicUsize::new(0));
|
77 + | let streams = Arc::new(AtomicUsize::new(0));
|
78 + | let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();
|
79 + |
|
80 + | let conns = connections.clone();
|
81 + | let strms = streams.clone();
|
82 + |
|
83 + | tokio::spawn(async move {
|
84 + | loop {
|
85 + | tokio::select! {
|
86 + | accept = listener.accept() => {
|
87 + | let (stream, _) = match accept {
|
88 + | Ok(v) => v,
|
89 + | Err(_) => break,
|
90 + | };
|
91 + | conns.fetch_add(1, Ordering::SeqCst);
|
92 + | let handler = handler.clone();
|
93 + | let strms = strms.clone();
|
94 + | tokio::spawn(async move {
|
95 + | let mut conn = h2::server::Builder::new()
|
96 + | .handshake(stream)
|
97 + | .await
|
98 + | .unwrap();
|
99 + | while let Some(result) = conn.accept().await {
|
100 + | let (req, respond) = result.unwrap();
|
101 + | strms.fetch_add(1, Ordering::SeqCst);
|
102 + | handler(req, respond);
|
103 + | }
|
104 + | });
|
105 + | }
|
106 + | _ = &mut shutdown_rx => break,
|
107 + | }
|
108 + | }
|
109 + | });
|
110 + |
|
111 + | Self {
|
112 + | addr,
|
113 + | connections,
|
114 + | streams,
|
115 + | _shutdown: shutdown_tx,
|
116 + | }
|
117 + | }
|
118 + |
|
119 + | /// Start a server that sends GOAWAY after `n` streams on each connection.
|
120 + | async fn start_goaway_after(n: usize, body: &'static str) -> Self {
|
121 + | let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
122 + | let addr = listener.local_addr().unwrap();
|
123 + | let connections = Arc::new(AtomicUsize::new(0));
|
124 + | let streams = Arc::new(AtomicUsize::new(0));
|
125 + | let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();
|
126 + |
|
127 + | let conns = connections.clone();
|
128 + | let strms = streams.clone();
|
129 + |
|
130 + | tokio::spawn(async move {
|
131 + | loop {
|
132 + | tokio::select! {
|
133 + | accept = listener.accept() => {
|
134 + | let (stream, _) = match accept {
|
135 + | Ok(v) => v,
|
136 + | Err(_) => break,
|
137 + | };
|
138 + | conns.fetch_add(1, Ordering::SeqCst);
|
139 + | let strms = strms.clone();
|
140 + | let per_conn_count = Arc::new(AtomicUsize::new(0));
|
141 + | tokio::spawn(async move {
|
142 + | let mut conn = h2::server::Builder::new()
|
143 + | .handshake(stream)
|
144 + | .await
|
145 + | .unwrap();
|
146 + | while let Some(result) = conn.accept().await {
|
147 + | let (req, mut respond) = result.unwrap();
|
148 + | strms.fetch_add(1, Ordering::SeqCst);
|
149 + | let count = per_conn_count.fetch_add(1, Ordering::SeqCst) + 1;
|
150 + |
|
151 + | // Respond normally
|
152 + | let response = http_1x::Response::builder()
|
153 + | .status(200)
|
154 + | .body(())
|
155 + | .unwrap();
|
156 + | let mut send_stream = respond.send_response(response, false).unwrap();
|
157 + | send_stream.send_data(Bytes::from(body), true).unwrap();
|
158 + | drop(req);
|
159 + |
|
160 + | // After n streams, send GOAWAY
|
161 + | if count >= n {
|
162 + | conn.graceful_shutdown();
|
163 + | }
|
164 + | }
|
165 + | });
|
166 + | }
|
167 + | _ = &mut shutdown_rx => break,
|
168 + | }
|
169 + | }
|
170 + | });
|
171 + |
|
172 + | Self {
|
173 + | addr,
|
174 + | connections,
|
175 + | streams,
|
176 + | _shutdown: shutdown_tx,
|
177 + | }
|
178 + | }
|
179 + |
|
180 + | fn connection_count(&self) -> usize {
|
181 + | self.connections.load(Ordering::SeqCst)
|
182 + | }
|
183 + |
|
184 + | fn stream_count(&self) -> usize {
|
185 + | self.streams.load(Ordering::SeqCst)
|
186 + | }
|
187 + |
|
188 + | fn url(&self) -> String {
|
189 + | format!("http://127.0.0.1:{}/", self.addr.port())
|
190 + | }
|
191 + | }
|
192 + |
|
193 + | // ---------------------------------------------------------------------------
|
194 + | // H2Connector — connects via TCP, signals negotiated_h2()
|
195 + | // ---------------------------------------------------------------------------
|
196 + |
|
197 + | /// IO wrapper that signals H2 negotiation to the pool's Negotiate layer.
|
198 + | struct H2Io {
|
199 + | inner: TcpStream,
|
200 + | }
|
201 + |
|
202 + | impl hyper_util::client::legacy::connect::Connection for H2Io {
|
203 + | fn connected(&self) -> Connected {
|
204 + | Connected::new().negotiated_h2()
|
205 + | }
|
206 + | }
|
207 + |
|
208 + | impl AsyncRead for H2Io {
|
209 + | fn poll_read(
|
210 + | mut self: Pin<&mut Self>,
|
211 + | cx: &mut Context<'_>,
|
212 + | buf: &mut ReadBuf<'_>,
|
213 + | ) -> Poll<std::io::Result<()>> {
|
214 + | Pin::new(&mut self.inner).poll_read(cx, buf)
|
215 + | }
|
216 + | }
|
217 + |
|
218 + | impl AsyncWrite for H2Io {
|
219 + | fn poll_write(
|
220 + | mut self: Pin<&mut Self>,
|
221 + | cx: &mut Context<'_>,
|
222 + | buf: &[u8],
|
223 + | ) -> Poll<std::io::Result<usize>> {
|
224 + | Pin::new(&mut self.inner).poll_write(cx, buf)
|
225 + | }
|
226 + |
|
227 + | fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
228 + | Pin::new(&mut self.inner).poll_flush(cx)
|
229 + | }
|
230 + |
|
231 + | fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
232 + | Pin::new(&mut self.inner).poll_shutdown(cx)
|
233 + | }
|
234 + | }
|
235 + |
|
236 + | // hyper::rt::Read and Write are needed by the pool
|
237 + | impl hyper::rt::Read for H2Io {
|
238 + | fn poll_read(
|
239 + | self: Pin<&mut Self>,
|
240 + | cx: &mut Context<'_>,
|
241 + | mut buf: hyper::rt::ReadBufCursor<'_>,
|
242 + | ) -> Poll<std::io::Result<()>> {
|
243 + | let n = unsafe {
|
244 + | let mut tbuf = ReadBuf::uninit(buf.as_mut());
|
245 + | match Pin::new(&mut self.get_mut().inner).poll_read(cx, &mut tbuf) {
|
246 + | Poll::Ready(Ok(())) => tbuf.filled().len(),
|
247 + | other => return other,
|
248 + | }
|
249 + | };
|
250 + | unsafe { buf.advance(n) };
|
251 + | Poll::Ready(Ok(()))
|
252 + | }
|
253 + | }
|
254 + |
|
255 + | impl hyper::rt::Write for H2Io {
|
256 + | fn poll_write(
|
257 + | mut self: Pin<&mut Self>,
|
258 + | cx: &mut Context<'_>,
|
259 + | buf: &[u8],
|
260 + | ) -> Poll<std::io::Result<usize>> {
|
261 + | Pin::new(&mut self.inner).poll_write(cx, buf)
|
262 + | }
|
263 + |
|
264 + | fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
265 + | Pin::new(&mut self.inner).poll_flush(cx)
|
266 + | }
|
267 + |
|
268 + | fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
269 + | Pin::new(&mut self.inner).poll_shutdown(cx)
|
270 + | }
|
271 + | }
|
272 + |
|
273 + | /// Connector that establishes TCP connections and signals H2 negotiation.
|
274 + | #[derive(Clone)]
|
275 + | struct H2Connector {
|
276 + | addr: SocketAddr,
|
277 + | }
|
278 + |
|
279 + | impl Service<http_1x::Uri> for H2Connector {
|
280 + | type Response = H2Io;
|
281 + | type Error = Box<dyn std::error::Error + Send + Sync>;
|
282 + | type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
|
283 + |
|
284 + | fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
285 + | Poll::Ready(Ok(()))
|
286 + | }
|
287 + |
|
288 + | fn call(&mut self, _req: http_1x::Uri) -> Self::Future {
|
289 + | let addr = self.addr;
|
290 + | Box::pin(async move {
|
291 + | let stream = TcpStream::connect(addr).await?;
|
292 + | Ok(H2Io { inner: stream })
|
293 + | })
|
294 + | }
|
295 + | }
|
296 + |
|
297 + | // ---------------------------------------------------------------------------
|
298 + | // Helpers
|
299 + | // ---------------------------------------------------------------------------
|
300 + |
|
301 + | fn build_h2_client(server: &H2MockServer) -> SharedHttpClient {
|
302 + | let pool =
|
303 + | SharedPool::builder().build_http_with_tcp_connector(H2Connector { addr: server.addr });
|
304 + | Client::new(&pool).into_shared()
|
305 + | }
|
306 + |
|
307 + | fn runtime_components() -> aws_smithy_runtime_api::client::runtime_components::RuntimeComponents {
|
308 + | RuntimeComponentsBuilder::for_tests()
|
309 + | .with_time_source(Some(aws_smithy_async::time::SystemTimeSource::new()))
|
310 + | .build()
|
311 + | .expect("valid runtime components")
|
312 + | }
|
313 + |
|
314 + | async fn send_request(
|
315 + | client: &SharedHttpClient,
|
316 + | url: &str,
|
317 + | ) -> Result<(u16, Vec<u8>), aws_smithy_runtime_api::client::result::ConnectorError> {
|
318 + | let settings = HttpConnectorSettings::builder().build();
|
319 + | let components = runtime_components();
|
320 + | let connector = client.http_connector(&settings, &components);
|
321 + | let resp = connector
|
322 + | .call(HttpRequest::get(url).expect("valid request"))
|
323 + | .await?;
|
324 + | let status = resp.status().as_u16();
|
325 + | let body = resp
|
326 + | .into_body()
|
327 + | .collect()
|
328 + | .await
|
329 + | .expect("body")
|
330 + | .to_bytes()
|
331 + | .to_vec();
|
332 + | Ok((status, body))
|
333 + | }
|
334 + |
|
335 + | // ---------------------------------------------------------------------------
|
336 + | // Tests
|
337 + | // ---------------------------------------------------------------------------
|
338 + |
|
339 + | /// Multiple concurrent requests multiplex over a single H2 connection.
|
340 + | #[tokio::test]
|
341 + | async fn h2_multiplexing_shares_one_connection() {
|
342 + | let server = H2MockServer::start("ok").await;
|
343 + | let client = build_h2_client(&server);
|
344 + | let url = server.url();
|
345 + |
|
346 + | // Warm the connection with one request first so the Singleton is populated
|
347 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
348 + | assert_eq!(status, 200);
|
349 + | assert_eq!(server.connection_count(), 1);
|
350 + |
|
351 + | // Now send 4 concurrent requests — they should all multiplex on the existing connection
|
352 + | let futs: Vec<_> = (0..4).map(|_| send_request(&client, &url)).collect();
|
353 + | let results = futures_util::future::join_all(futs).await;
|
354 + |
|
355 + | for (i, r) in results.iter().enumerate() {
|
356 + | let (status, _) = r
|
357 + | .as_ref()
|
358 + | .unwrap_or_else(|e| panic!("request {i} failed: {e}"));
|
359 + | assert_eq!(*status, 200);
|
360 + | }
|
361 + |
|
362 + | // All 5 requests (1 warm + 4 concurrent) should have used a single connection
|
363 + | assert_eq!(
|
364 + | server.connection_count(),
|
365 + | 1,
|
366 + | "H2 should multiplex on one connection"
|
367 + | );
|
368 + | assert_eq!(server.stream_count(), 5, "should have 5 streams");
|
369 + | }
|
370 + |
|
371 + | /// After GOAWAY, the pool establishes a new connection for subsequent requests.
|
372 + | #[tokio::test]
|
373 + | async fn h2_goaway_triggers_new_connection() {
|
374 + | // Server sends GOAWAY after 2 streams per connection
|
375 + | let server = H2MockServer::start_goaway_after(2, "ok").await;
|
376 + | let client = build_h2_client(&server);
|
377 + | let url = server.url();
|
378 + |
|
379 + | // First 2 requests on connection 1
|
380 + | let (s1, _) = send_request(&client, &url).await.unwrap();
|
381 + | let (s2, _) = send_request(&client, &url).await.unwrap();
|
382 + | assert_eq!(s1, 200);
|
383 + | assert_eq!(s2, 200);
|
384 + |
|
385 + | // Give the pool time to observe the GOAWAY
|
386 + | tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
387 + |
|
388 + | // Next request should go on a new connection
|
389 + | let (s3, _) = send_request(&client, &url).await.unwrap();
|
390 + | assert_eq!(s3, 200);
|
391 + |
|
392 + | assert!(
|
393 + | server.connection_count() >= 2,
|
394 + | "should have opened a second connection after GOAWAY, got {}",
|
395 + | server.connection_count()
|
396 + | );
|
397 + | }
|
398 + |
|
399 + | /// Sequential requests reuse the same H2 connection (no unnecessary reconnects).
|
400 + | #[tokio::test]
|
401 + | async fn h2_sequential_requests_reuse_connection() {
|
402 + | let server = H2MockServer::start("hello").await;
|
403 + | let client = build_h2_client(&server);
|
404 + | let url = server.url();
|
405 + |
|
406 + | for i in 0..5 {
|
407 + | let (status, body) = send_request(&client, &url).await.unwrap();
|
408 + | assert_eq!(status, 200, "request {i}");
|
409 + | assert_eq!(body, b"hello", "request {i}");
|
410 + | }
|
411 + |
|
412 + | assert_eq!(
|
413 + | server.connection_count(),
|
414 + | 1,
|
415 + | "should reuse one H2 connection"
|
416 + | );
|
417 + | assert_eq!(server.stream_count(), 5);
|
418 + | }
|
419 + |
|
420 + | /// Poisoning an H2 connection forces the pool to establish a new one.
|
421 + | #[tokio::test]
|
422 + | async fn h2_poisoned_connection_not_reused() {
|
423 + | use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
|
424 + |
|
425 + | let server = H2MockServer::start("ok").await;
|
426 + | let client = build_h2_client(&server);
|
427 + | let url = server.url();
|
428 + |
|
429 + | // First request: establish H2 connection, capture metadata.
|
430 + | let settings = HttpConnectorSettings::builder().build();
|
431 + | let components = runtime_components();
|
432 + | let connector = client.http_connector(&settings, &components);
|
433 + |
|
434 + | let capture = CaptureSmithyConnection::new();
|
435 + | let mut request = HttpRequest::get(&url).expect("valid request");
|
436 + | request.add_extension(capture.clone());
|
437 + |
|
438 + | let resp = connector
|
439 + | .call(request)
|
440 + | .await
|
441 + | .expect("request should succeed");
|
442 + | let _body = resp.into_body().collect().await.expect("body");
|
443 + | assert_eq!(server.connection_count(), 1);
|
444 + |
|
445 + | let metadata = capture.get().expect("adapter should populate metadata");
|
446 + | metadata.poison();
|
447 + |
|
448 + | // Give the pool a moment to observe the poison
|
449 + | tokio::time::sleep(std::time::Duration::from_millis(20)).await;
|
450 + |
|
451 + | // Next request should open a NEW H2 connection
|
452 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
453 + | assert_eq!(status, 200);
|
454 + | assert!(
|
455 + | server.connection_count() >= 2,
|
456 + | "poisoned H2 connection should not be reused, got {} connections",
|
457 + | server.connection_count()
|
458 + | );
|
459 + | }
|
460 + |
|
461 + | // ===========================================================================
|
462 + | // TLS + ALPN — real certificate negotiation
|
463 + | // ===========================================================================
|
464 + | //
|
465 + | // These tests use a TLS server with self-signed certs advertising h2 via ALPN.
|
466 + | // They verify the v2 pool correctly routes to H2 when ALPN negotiates it
|
467 + | // through real TLS.
|
468 + |
|
469 + | #[cfg(feature = "rustls-aws-lc")]
|
470 + | mod tls_h2 {
|
471 + | use aws_smithy_http_client::pool::{Client, SharedPool};
|
472 + | use aws_smithy_http_client::tls;
|
473 + | use aws_smithy_http_client::tls::{TlsContext, TrustStore};
|
474 + | use aws_smithy_runtime_api::client::http::{
|
475 + | HttpClient, HttpConnector, HttpConnectorSettings, SharedHttpClient,
|
476 + | };
|
477 + | use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
|
478 + | use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
|
479 + | use aws_smithy_runtime_api::shared::IntoShared;
|
480 + | use http_body_util::BodyExt;
|
481 + | use hyper_util::rt::TokioExecutor;
|
482 + | use std::net::SocketAddr;
|
483 + | use std::sync::atomic::{AtomicUsize, Ordering};
|
484 + | use std::sync::Arc;
|
485 + | use tokio::net::TcpListener;
|
486 + | use tokio_rustls::TlsAcceptor;
|
487 + |
|
488 + | /// TLS test server that tracks connection count and serves H2 via ALPN.
|
489 + | struct TlsH2Server {
|
490 + | addr: SocketAddr,
|
491 + | connections: Arc<AtomicUsize>,
|
492 + | _shutdown: tokio::sync::oneshot::Sender<()>,
|
493 + | }
|
494 + |
|
495 + | impl TlsH2Server {
|
496 + | async fn start() -> Self {
|
497 + | let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
|
498 + |
|
499 + | let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
500 + | let addr = listener.local_addr().unwrap();
|
501 + | let connections = Arc::new(AtomicUsize::new(0));
|
502 + | let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();
|
503 + |
|
504 + | // Load certs
|
505 + | let cert_pem = std::fs::read("tests/server.pem").unwrap();
|
506 + | let key_pem = std::fs::read("tests/server.rsa").unwrap();
|
507 + | let certs: Vec<_> = rustls_pemfile::certs(&mut &cert_pem[..])
|
508 + | .collect::<Result<_, _>>()
|
509 + | .unwrap();
|
510 + | let key = rustls_pemfile::private_key(&mut &key_pem[..])
|
511 + | .unwrap()
|
512 + | .unwrap();
|
513 + |
|
514 + | let mut server_config = rustls::ServerConfig::builder()
|
515 + | .with_no_client_auth()
|
516 + | .with_single_cert(certs, key)
|
517 + | .unwrap();
|
518 + | // Only advertise h2 — force H2 negotiation
|
519 + | server_config.alpn_protocols = vec![b"h2".to_vec()];
|
520 + | let tls_acceptor = TlsAcceptor::from(Arc::new(server_config));
|
521 + |
|
522 + | let conns = connections.clone();
|
523 + | tokio::spawn(async move {
|
524 + | loop {
|
525 + | tokio::select! {
|
526 + | accept = listener.accept() => {
|
527 + | let (tcp, _) = match accept {
|
528 + | Ok(v) => v,
|
529 + | Err(_) => break,
|
530 + | };
|
531 + | conns.fetch_add(1, Ordering::SeqCst);
|
532 + | let tls_acceptor = tls_acceptor.clone();
|
533 + | tokio::spawn(async move {
|
534 + | let tls_stream = match tls_acceptor.accept(tcp).await {
|
535 + | Ok(s) => s,
|
536 + | Err(e) => {
|
537 + | eprintln!("TLS accept failed: {e}");
|
538 + | return;
|
539 + | }
|
540 + | };
|
541 + | let service = hyper::service::service_fn(|_req| async {
|
542 + | Ok::<_, hyper::Error>(
|
543 + | http_1x::Response::builder()
|
544 + | .status(200)
|
545 + | .body(http_body_util::Full::new(
|
546 + | bytes::Bytes::from("h2-ok"),
|
547 + | ))
|
548 + | .unwrap(),
|
549 + | )
|
550 + | });
|
551 + | let io = hyper_util::rt::TokioIo::new(tls_stream);
|
552 + | // Use http2 only server since we only advertise h2
|
553 + | let _ = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new())
|
554 + | .serve_connection(io, service)
|
555 + | .await;
|
556 + | });
|
557 + | }
|
558 + | _ = &mut shutdown_rx => break,
|
559 + | }
|
560 + | }
|
561 + | });
|
562 + |
|
563 + | Self {
|
564 + | addr,
|
565 + | connections,
|
566 + | _shutdown: shutdown_tx,
|
567 + | }
|
568 + | }
|
569 + |
|
570 + | fn connection_count(&self) -> usize {
|
571 + | self.connections.load(Ordering::SeqCst)
|
572 + | }
|
573 + |
|
574 + | fn url(&self) -> String {
|
575 + | format!("https://localhost:{}/", self.addr.port())
|
576 + | }
|
577 + | }
|
578 + |
|
579 + | fn tls_context() -> TlsContext {
|
580 + | let pem = std::fs::read("tests/server.pem").unwrap();
|
581 + | let trust_store = TrustStore::empty().with_pem_certificate(pem);
|
582 + | TlsContext::builder()
|
583 + | .with_trust_store(trust_store)
|
584 + | .build()
|
585 + | .unwrap()
|
586 + | }
|
587 + |
|
588 + | fn runtime_components() -> aws_smithy_runtime_api::client::runtime_components::RuntimeComponents
|
589 + | {
|
590 + | RuntimeComponentsBuilder::for_tests()
|
591 + | .with_time_source(Some(aws_smithy_async::time::SystemTimeSource::new()))
|
592 + | .build()
|
593 + | .unwrap()
|
594 + | }
|
595 + |
|
596 + | async fn send_request(
|
597 + | client: &SharedHttpClient,
|
598 + | url: &str,
|
599 + | ) -> Result<(u16, Vec<u8>), aws_smithy_runtime_api::client::result::ConnectorError> {
|
600 + | let settings = HttpConnectorSettings::builder().build();
|
601 + | let components = runtime_components();
|
602 + | let connector = client.http_connector(&settings, &components);
|
603 + | let resp = connector
|
604 + | .call(HttpRequest::get(url).expect("valid request"))
|
605 + | .await?;
|
606 + | let status = resp.status().as_u16();
|
607 + | let body = resp
|
608 + | .into_body()
|
609 + | .collect()
|
610 + | .await
|
611 + | .expect("body")
|
612 + | .to_bytes()
|
613 + | .to_vec();
|
614 + | Ok((status, body))
|
615 + | }
|
616 + |
|
617 + | /// Rustls + ALPN h2: v2 pool routes to H2, multiple requests multiplex.
|
618 + | #[tokio::test]
|
619 + | async fn rustls_alpn_h2_multiplexing() {
|
620 + | let server = TlsH2Server::start().await;
|
621 + | let pool = SharedPool::builder()
|
622 + | .tls_provider(tls::Provider::Rustls(
|
623 + | tls::rustls_provider::CryptoMode::AwsLc,
|
624 + | ))
|
625 + | .tls_context(tls_context())
|
626 + | .build_https();
|
627 + | let client: SharedHttpClient = Client::new(&pool).into_shared();
|
628 + |
|
629 + | let url = server.url();
|
630 + |
|
631 + | // First request establishes connection
|
632 + | let (status, body) = send_request(&client, &url).await.unwrap();
|
633 + | assert_eq!(status, 200);
|
634 + | assert_eq!(body, b"h2-ok");
|
635 + |
|
636 + | // Second request should multiplex on same connection
|
637 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
638 + | assert_eq!(status, 200);
|
639 + |
|
640 + | // Third request too
|
641 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
642 + | assert_eq!(status, 200);
|
643 + |
|
644 + | assert_eq!(
|
645 + | server.connection_count(),
|
646 + | 1,
|
647 + | "rustls H2: all requests should multiplex on one connection"
|
648 + | );
|
649 + | }
|
650 + |
|
651 + | /// s2n-tls + ALPN h2: the v2 pool routes to H2 and multiplexes. Three
|
652 + | /// sequential requests share one connection because `S2nTlsConn::connected()`
|
653 + | /// reports the negotiated protocol from the TLS `application_protocol`, so the
|
654 + | /// Negotiate layer selects the H2 leg.
|
655 + | #[cfg(feature = "s2n-tls")]
|
656 + | #[tokio::test]
|
657 + | async fn s2n_tls_alpn_h2_multiplexing() {
|
658 + | let server = TlsH2Server::start().await;
|
659 + | let pool = SharedPool::builder()
|
660 + | .tls_provider(tls::Provider::S2nTls)
|
661 + | .tls_context(tls_context())
|
662 + | .build_https();
|
663 + | let client: SharedHttpClient = Client::new(&pool).into_shared();
|
664 + |
|
665 + | let url = server.url();
|
666 + |
|
667 + | // First request
|
668 + | let (status, body) = send_request(&client, &url).await.unwrap();
|
669 + | assert_eq!(status, 200);
|
670 + | assert_eq!(body, b"h2-ok");
|
671 + |
|
672 + | // Second request — if H2 works, should multiplex on same connection
|
673 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
674 + | assert_eq!(status, 200);
|
675 + |
|
676 + | // Third request
|
677 + | let (status, _) = send_request(&client, &url).await.unwrap();
|
678 + | assert_eq!(status, 200);
|
679 + |
|
680 + | // All three requests multiplex on one H2 connection.
|
681 + | assert_eq!(
|
682 + | server.connection_count(),
|
683 + | 1,
|
684 + | "s2n-tls H2: all requests should multiplex on one connection"
|
685 + | );
|
686 + | }
|
687 + | }
|