diff --git a/Cargo.lock b/Cargo.lock index 119a8b3afed2c..d1d2585208773 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1465,6 +1465,33 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" +[[package]] +name = "aws-lc-rs" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f95446d919226d587817a7d21379e6eb099b97b45110a7f272a444ca5c54070" +dependencies = [ + "aws-lc-sys", + "mirai-annotations", + "paste", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.21.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3ddc4a5b231dd6958b140ff3151b6412b3f4321fab354f399eec8f14b06df62" +dependencies = [ + "bindgen 0.69.4", + "cc", + "cmake", + "dunce", + "fs_extra", + "libc", + "paste", +] + [[package]] name = "backoff" version = "0.4.0" @@ -1615,6 +1642,29 @@ dependencies = [ "syn 2.0.79", ] +[[package]] +name = "bindgen" +version = "0.69.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00dc851838a2120612785d195287475a3ac45514741da670b735818822129a0" +dependencies = [ + "bitflags 2.6.0", + "cexpr", + "clang-sys", + "itertools 0.11.0", + "lazy_static", + "lazycell", + "log", + "prettyplease", + "proc-macro2 1.0.86", + "quote 1.0.37", + "regex", + "rustc-hash 1.1.0", + "shlex", + "syn 2.0.79", + "which", +] + [[package]] name = "bip39" version = "2.0.0" @@ -2708,12 +2758,13 @@ checksum = "a2698f953def977c68f935bb0dfa959375ad4638570e969e2f1e9f433cbf1af6" [[package]] name = "cc" -version = "1.0.94" +version = "1.1.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17f6e324229dc011159fcc089755d1e2e216a90d43a7dea6853ca740b84f35e7" +checksum = "3bbb537bb4a30b90362caddba8f360c0a56bc13d3a5570028e7197204cb54a17" dependencies = [ "jobserver", "libc", + "shlex", ] [[package]] @@ -3049,6 +3100,15 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "98cc8fbded0c607b7ba9dd60cd98df59af97e84d24e49c8557331cfc26d301ce" +[[package]] +name = "cmake" +version = "0.1.51" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb1e43aa7fd152b1f968787f7dbcdeb306d1867ff373c69955211876c053f91a" +dependencies = [ + "cc", +] + [[package]] name = "coarsetime" version = "0.1.23" @@ -7430,9 +7490,9 @@ dependencies = [ [[package]] name = "hyper-rustls" -version = "0.27.2" +version = "0.27.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ee4be2c948921a1a5320b629c4193916ed787a7f7f293fd3f7f5a6c9de74155" +checksum = "08afdbb5c31130e3034af566421053ab03787c640246a446327f550d11bcb333" dependencies = [ "futures-util", "http 1.1.0", @@ -7440,6 +7500,7 @@ dependencies = [ "hyper-util", "log", "rustls 0.23.10", + "rustls-native-certs 0.8.0", "rustls-pki-types", "tokio", "tokio-rustls 0.26.0", @@ -7918,9 +7979,9 @@ checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130" [[package]] name = "jobserver" -version = "0.1.26" +version = "0.1.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "936cfd212a0155903bcbc060e316fb6cc7cbf2e1907329391ebadc1fe0ce77c2" +checksum = "48d1dbcbbeb6a7fec7e059840aa538bd62aaccf972c7346c4d9d2059312853d0" dependencies = [ "libc", ] @@ -8182,7 +8243,7 @@ dependencies = [ "base64 0.22.1", "http-body 1.0.0", "hyper 1.3.1", - "hyper-rustls 0.27.2", + "hyper-rustls 0.27.3", "hyper-util", "jsonrpsee-core 0.24.3", "jsonrpsee-types 0.24.3", @@ -9048,7 +9109,7 @@ version = "0.11.0+8.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3386f101bcb4bd252d8e9d2fb41ec3b0862a15a62b478c355b2982efa469e3e" dependencies = [ - "bindgen", + "bindgen 0.65.1", "bzip2-sys", "cc", "glob", @@ -9616,6 +9677,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "mirai-annotations" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9be0862c1b3f26a88803c4a49de6889c10e608b3ee9344e6ef5b45fb37ad3d1" + [[package]] name = "mixnet" version = "0.7.0" @@ -17519,7 +17586,7 @@ dependencies = [ "http-body 1.0.0", "http-body-util", "hyper 1.3.1", - "hyper-rustls 0.27.2", + "hyper-rustls 0.27.3", "hyper-util", "ipnet", "js-sys", @@ -18099,6 +18166,7 @@ version = "0.23.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05cff451f60db80f490f3c182b77c35260baace73209e9cdbbe526bfe3a4d402" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring 0.17.7", @@ -18133,6 +18201,19 @@ dependencies = [ "security-framework", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcaf18a4f2be7326cd874a5fa579fae794320a0f388d365dca7e480e55f83f8a" +dependencies = [ + "openssl-probe", + "rustls-pemfile 2.0.0", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pemfile" version = "1.0.3" @@ -18201,6 +18282,7 @@ version = "0.102.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff448f7e92e913c4b7d4c6d8e4540a1724b319b4152b8aef6d4cf8339712b33e" dependencies = [ + "aws-lc-rs", "ring 0.17.7", "rustls-pki-types", "untrusted 0.9.0", @@ -19373,8 +19455,10 @@ dependencies = [ "fnv", "futures", "futures-timer", - "hyper 0.14.29", - "hyper-rustls 0.24.2", + "http-body-util", + "hyper 1.3.1", + "hyper-rustls 0.27.3", + "hyper-util", "log", "num_cpus", "once_cell", @@ -26441,6 +26525,18 @@ dependencies = [ "westend-emulated-chain", ] +[[package]] +name = "which" +version = "4.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7" +dependencies = [ + "either", + "home", + "once_cell", + "rustix 0.38.21", +] + [[package]] name = "wide" version = "0.7.11" diff --git a/Cargo.toml b/Cargo.toml index df662ac269880..771542007cb45 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -797,10 +797,8 @@ http = { version = "1.1" } http-body = { version = "1", default-features = false } http-body-util = { version = "0.1.2", default-features = false } hyper = { version = "1.3.1", default-features = false } -hyper-rustls = { version = "0.24.2" } +hyper-rustls = { version = "0.27.3" } hyper-util = { version = "0.1.5", default-features = false } -# TODO: remove hyper v0.14 https://github.com/paritytech/polkadot-sdk/issues/4896 -hyperv14 = { package = "hyper", version = "0.14.29", default-features = false } impl-serde = { version = "0.4.0", default-features = false } impl-trait-for-tuples = { version = "0.2.2" } indexmap = { version = "2.0.0" } diff --git a/substrate/client/offchain/Cargo.toml b/substrate/client/offchain/Cargo.toml index bbbe7018d1060..49488589b2d59 100644 --- a/substrate/client/offchain/Cargo.toml +++ b/substrate/client/offchain/Cargo.toml @@ -22,11 +22,10 @@ codec = { features = ["derive"], workspace = true, default-features = true } fnv = { workspace = true } futures = { workspace = true } futures-timer = { workspace = true } -hyperv14 = { features = [ - "http2", - "stream", -], workspace = true, default-features = true } +http-body-util = { workspace = true } +hyper = { features = ["http2"], workspace = true, default-features = true } hyper-rustls = { features = ["http2"], workspace = true } +hyper-util = { features = ["client-legacy"], workspace = true } num_cpus = { workspace = true } once_cell = { workspace = true } parking_lot = { workspace = true, default-features = true } diff --git a/substrate/client/offchain/src/api/http.rs b/substrate/client/offchain/src/api/http.rs index 73407b1359d77..f5699db685b0c 100644 --- a/substrate/client/offchain/src/api/http.rs +++ b/substrate/client/offchain/src/api/http.rs @@ -27,14 +27,17 @@ //! (i.e.: the socket should continue being processed) in the background even if the runtime isn't //! actively calling any function. -use hyperv14 as hyper; - use crate::api::timestamp; use bytes::buf::{Buf, Reader}; use fnv::FnvHashMap; use futures::{channel::mpsc, future, prelude::*}; -use hyper::{client, Body, Client as HyperClient}; +use http_body_util::{combinators::BoxBody, StreamBody}; +use hyper::body::Body as _; use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder}; +use hyper_util::{ + client::legacy::{self as client, Client as HyperClient}, + rt::TokioExecutor, +}; use once_cell::sync::Lazy; use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender}; use sp_core::offchain::{HttpError, HttpRequestId, HttpRequestStatus, Timestamp}; @@ -48,20 +51,28 @@ use std::{ const LOG_TARGET: &str = "offchain-worker::http"; +pub type Body = BoxBody; + +type Sender = mpsc::Sender, hyper::Error>>; +type Receiver = mpsc::Receiver, hyper::Error>>; + /// Wrapper struct used for keeping the hyper_rustls client running. #[derive(Clone)] -pub struct SharedClient(Arc, Body>>>); +pub struct SharedClient( + Arc, Body>>>, +); impl SharedClient { pub fn new() -> Self { Self(Arc::new(Lazy::new(|| { let connector = HttpsConnectorBuilder::new() .with_native_roots() + .expect("Failed to build HttpsConnector with native roots") .https_or_http() .enable_http1() .enable_http2() .build(); - HyperClient::builder().build(connector) + HyperClient::builder(TokioExecutor::new()).build(connector) }))) } } @@ -105,23 +116,23 @@ pub struct HttpApi { /// One active request within `HttpApi`. enum HttpApiRequest { /// The request object is being constructed locally and not started yet. - NotDispatched(hyper::Request, hyper::body::Sender), + NotDispatched(hyper::Request, Sender), /// The request has been dispatched and we're in the process of sending out the body (if the /// field is `Some`) or waiting for a response (if the field is `None`). - Dispatched(Option), + Dispatched(Option), /// Received a response. Response(HttpApiRequestRp), /// A request has been dispatched but the worker notified us of an error. We report this /// failure to the user as an `IoError` and remove the request from the list as soon as /// possible. - Fail(hyper::Error), + Fail(client::Error), } /// A request within `HttpApi` that has received a response. struct HttpApiRequestRp { /// We might still be writing the request's body when the response comes. /// This field allows to continue writing that body. - sending_body: Option, + sending_body: Option, /// Status code of the response. status_code: hyper::StatusCode, /// Headers of the response. @@ -132,7 +143,7 @@ struct HttpApiRequestRp { /// Elements extracted from the channel are first put into `current_read_chunk`. /// If the channel produces an error, then that is translated into an `IoError` and the request /// is removed from the list. - body: stream::Fuse>>, + body: stream::Fuse, /// Chunk that has been extracted from the channel and that is currently being read. /// Reading data from the response should read from this field in priority. current_read_chunk: Option>, @@ -144,7 +155,9 @@ impl HttpApi { // Start by building the prototype of the request. // We do this first so that we don't touch anything in `self` if building the prototype // fails. - let (body_sender, body) = hyper::Body::channel(); + let (body_sender, receiver) = mpsc::channel(0); + let body = StreamBody::new(receiver); + let body = BoxBody::new(body); let mut request = hyper::Request::new(body); *request.method_mut() = hyper::Method::from_bytes(method.as_bytes()).map_err(|_| ())?; *request.uri_mut() = hyper::Uri::from_maybe_shared(uri.to_owned()).map_err(|_| ())?; @@ -158,7 +171,7 @@ impl HttpApi { target: LOG_TARGET, "Overflow in offchain worker HTTP request ID assignment" ); - return Err(()) + return Err(()); }, }; self.requests @@ -213,7 +226,7 @@ impl HttpApi { // Closure that writes data to a sender, taking the deadline into account. Can return `Ok` // (if the body has been written), or `DeadlineReached`, or `IoError`. // If `IoError` is returned, don't forget to remove the request from the list. - let mut poll_sender = move |sender: &mut hyper::body::Sender| -> Result<(), HttpError> { + let mut poll_sender = move |sender: &mut Sender| -> Result<(), HttpError> { let mut when_ready = future::maybe_done(future::poll_fn(|cx| sender.poll_ready(cx))); futures::executor::block_on(future::select(&mut when_ready, &mut deadline)); match when_ready { @@ -221,12 +234,15 @@ impl HttpApi { future::MaybeDone::Done(Err(_)) => return Err(HttpError::IoError), future::MaybeDone::Future(_) | future::MaybeDone::Gone => { debug_assert!(matches!(deadline, future::MaybeDone::Done(..))); - return Err(HttpError::DeadlineReached) + return Err(HttpError::DeadlineReached); }, }; futures::executor::block_on( - sender.send_data(hyper::body::Bytes::from(chunk.to_owned())), + async { + future::poll_fn(|cx| sender.poll_ready(cx)).await?; + sender.start_send(Ok(hyper::body::Frame::data(hyper::body::Bytes::from(chunk.to_owned())))) + } ) .map_err(|_| { tracing::error!(target: "offchain-worker::http", "HTTP sender refused data despite being ready"); @@ -250,13 +266,13 @@ impl HttpApi { match poll_sender(&mut sender) { Err(HttpError::IoError) => { tracing::debug!(target: LOG_TARGET, id = %request_id.0, "Encountered io error while trying to add new chunk to body"); - return Err(HttpError::IoError) + return Err(HttpError::IoError); }, other => { tracing::debug!(target: LOG_TARGET, id = %request_id.0, res = ?other, "Added chunk to body"); self.requests .insert(request_id, HttpApiRequest::Dispatched(Some(sender))); - return other + return other; }, } } else { @@ -265,7 +281,7 @@ impl HttpApi { // Writing an empty body is a hint that we should stop writing. Dropping // the sender. self.requests.insert(request_id, HttpApiRequest::Dispatched(None)); - return Ok(()) + return Ok(()); } }, @@ -281,13 +297,13 @@ impl HttpApi { ) { Err(HttpError::IoError) => { tracing::debug!(target: LOG_TARGET, id = %request_id.0, "Encountered io error while trying to add new chunk to body"); - return Err(HttpError::IoError) + return Err(HttpError::IoError); }, other => { tracing::debug!(target: LOG_TARGET, id = %request_id.0, res = ?other, "Added chunk to body"); self.requests .insert(request_id, HttpApiRequest::Response(response)); - return other + return other; }, } } else { @@ -302,7 +318,7 @@ impl HttpApi { ..response }), ); - return Ok(()) + return Ok(()); } }, @@ -311,7 +327,7 @@ impl HttpApi { // If the request has already failed, return without putting back the request // in the list. - return Err(HttpError::IoError) + return Err(HttpError::IoError); }, v @ HttpApiRequest::Dispatched(None) | @@ -320,7 +336,7 @@ impl HttpApi { // We have already finished sending this body. self.requests.insert(request_id, v); - return Err(HttpError::Invalid) + return Err(HttpError::Invalid); }, } } @@ -340,7 +356,7 @@ impl HttpApi { Some(HttpApiRequest::Dispatched(sending_body)) | Some(HttpApiRequest::Response(HttpApiRequestRp { sending_body, .. })) => { let _ = sending_body.take(); - continue + continue; }, _ => continue, }; @@ -405,7 +421,7 @@ impl HttpApi { }, } } - return output + return output; } } @@ -418,7 +434,7 @@ impl HttpApi { msg } else { debug_assert!(matches!(deadline, future::MaybeDone::Done(..))); - continue + continue; } }; @@ -458,7 +474,7 @@ impl HttpApi { None => { tracing::error!(target: "offchain-worker::http", "Worker has crashed"); - return ids.iter().map(|_| HttpRequestStatus::IoError).collect() + return ids.iter().map(|_| HttpRequestStatus::IoError).collect(); }, } } @@ -498,14 +514,14 @@ impl HttpApi { // and we still haven't received a response. Some(rq @ HttpApiRequest::Dispatched(_)) => { self.requests.insert(request_id, rq); - return Err(HttpError::DeadlineReached) + return Err(HttpError::DeadlineReached); }, // The request has failed. Some(HttpApiRequest::Fail { .. }) => return Err(HttpError::IoError), // Request hasn't been dispatched yet; reading the body is invalid. Some(rq @ HttpApiRequest::NotDispatched(_, _)) => { self.requests.insert(request_id, rq); - return Err(HttpError::Invalid) + return Err(HttpError::Invalid); }, None => return Err(HttpError::Invalid), }; @@ -526,12 +542,12 @@ impl HttpApi { ..response }), ); - return Ok(n) + return Ok(n); }, Err(err) => { // This code should never be reached unless there's a logic error somewhere. tracing::error!(target: "offchain-worker::http", "Failed to read from current read chunk: {:?}", err); - return Err(HttpError::IoError) + return Err(HttpError::IoError); }, } } @@ -544,7 +560,10 @@ impl HttpApi { if let future::MaybeDone::Done(next_body) = next_body { match next_body { - Some(Ok(chunk)) => response.current_read_chunk = Some(chunk.reader()), + Some(Ok(chunk)) => + if let Ok(chunk) = chunk.into_data() { + response.current_read_chunk = Some(chunk.reader()); + }, Some(Err(_)) => return Err(HttpError::IoError), None => return Ok(0), // eof } @@ -552,7 +571,7 @@ impl HttpApi { if let future::MaybeDone::Done(_) = deadline { self.requests.insert(request_id, HttpApiRequest::Response(response)); - return Err(HttpError::DeadlineReached) + return Err(HttpError::DeadlineReached); } } } @@ -587,7 +606,7 @@ enum ApiToWorker { /// ID to send back when the response comes back. id: HttpRequestId, /// Request to start executing. - request: hyper::Request, + request: hyper::Request, }, } @@ -608,14 +627,14 @@ enum WorkerToApi { /// the next item. /// Can also be used to send an error, in case an error happened on the HTTP socket. After /// an error is sent, the channel will close. - body: mpsc::Receiver>, + body: Receiver, }, /// A request has failed because of an error. The request is then no longer valid. Fail { /// The ID that was passed to the worker. id: HttpRequestId, /// Error that happened. - error: hyper::Error, + error: client::Error, }, } @@ -626,7 +645,7 @@ pub struct HttpWorker { /// Used to receive messages from the `HttpApi`. from_api: TracingUnboundedReceiver, /// The engine that runs HTTP requests. - http_client: Arc, Body>>>, + http_client: Arc, Body>>>, /// HTTP requests that are being worked on by the engine. requests: Vec<(HttpRequestId, HttpWorkerRequest)>, } @@ -634,13 +653,13 @@ pub struct HttpWorker { /// HTTP request being processed by the worker. enum HttpWorkerRequest { /// Request has been dispatched and is waiting for a response from the Internet. - Dispatched(hyper::client::ResponseFuture), + Dispatched(client::ResponseFuture), /// Progressively reading the body of the response and sending it to the channel. ReadBody { /// Body to read `Chunk`s from. Only used if the channel is ready to accept data. - body: hyper::Body, + body: Body, /// Channel to the [`HttpApi`] where we send the chunks to. - tx: mpsc::Sender>, + tx: Sender, }, } @@ -663,12 +682,12 @@ impl Future for HttpWorker { let response = match Future::poll(Pin::new(&mut future), cx) { Poll::Pending => { me.requests.push((id, HttpWorkerRequest::Dispatched(future))); - continue + continue; }, Poll::Ready(Ok(response)) => response, Poll::Ready(Err(error)) => { let _ = me.to_api.unbounded_send(WorkerToApi::Fail { id, error }); - continue // don't insert the request back + continue; // don't insert the request back }, }; @@ -684,9 +703,12 @@ impl Future for HttpWorker { body: body_rx, }); - me.requests.push((id, HttpWorkerRequest::ReadBody { body, tx: body_tx })); + me.requests.push(( + id, + HttpWorkerRequest::ReadBody { body: Body::new(body), tx: body_tx }, + )); cx.waker().wake_by_ref(); // reschedule in order to poll the new future - continue + continue; }, HttpWorkerRequest::ReadBody { mut body, mut tx } => { @@ -697,12 +719,11 @@ impl Future for HttpWorker { Poll::Ready(Err(_)) => continue, // don't insert the request back Poll::Pending => { me.requests.push((id, HttpWorkerRequest::ReadBody { body, tx })); - continue + continue; }, } - // `tx` is ready. Read a chunk from the socket and send it to the channel. - match Stream::poll_next(Pin::new(&mut body), cx) { + match Pin::new(&mut body).poll_frame(cx) { Poll::Ready(Some(Ok(chunk))) => { let _ = tx.start_send(Ok(chunk)); me.requests.push((id, HttpWorkerRequest::ReadBody { body, tx })); @@ -762,7 +783,8 @@ mod tests { }; use crate::api::timestamp; use core::convert::Infallible; - use futures::{future, StreamExt}; + use futures::future; + use http_body_util::BodyExt; use sp_core::offchain::{Duration, Externalities, HttpError, HttpRequestId, HttpRequestStatus}; use std::sync::LazyLock; @@ -774,7 +796,9 @@ mod tests { // server that runs in the background as well. macro_rules! build_api_server { () => { - build_api_server!(hyper::Response::new(hyper::Body::from("Hello World!"))) + build_api_server!(hyper::Response::new(http_body_util::Full::new( + hyper::body::Bytes::from("Hello World!") + ))) }; ( $response:expr ) => {{ let hyper_client = SHARED_CLIENT.clone(); @@ -785,21 +809,32 @@ mod tests { let rt = tokio::runtime::Runtime::new().unwrap(); let worker = rt.spawn(worker); let server = rt.spawn(async move { - let server = hyper::Server::bind(&"127.0.0.1:0".parse().unwrap()).serve( - hyper::service::make_service_fn(|_| async move { - Ok::<_, Infallible>(hyper::service::service_fn( - move |req: hyper::Request| async move { - // Wait until the complete request was received and processed, - // otherwise the tests are flaky. - let _ = req.into_body().collect::>().await; - - Ok::<_, Infallible>($response) - }, - )) - }), - ); - let _ = addr_tx.send(server.local_addr()); - server.await.map_err(drop) + let addr = std::net::SocketAddr::from(([127, 0, 0, 1], 0)); + let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); + let _ = addr_tx.send(listener.local_addr().unwrap()); + loop { + let (stream, _) = listener.accept().await.unwrap(); + let io = hyper_util::rt::TokioIo::new(stream); + tokio::task::spawn(async move { + if let Err(err) = hyper::server::conn::http1::Builder::new() + .serve_connection( + io, + hyper::service::service_fn( + move |req: hyper::Request| async move { + // Wait until the complete request was received and + // processed, otherwise the tests are flaky. + let _ = req.into_body().collect().await; + + Ok::<_, Infallible>($response) + }, + ), + ) + .await + { + eprintln!("Error serving connection: {:?}", err); + } + }); + } }); let _ = rt.block_on(future::join(worker, server)); }); @@ -839,7 +874,7 @@ mod tests { let (mut api, addr) = build_api_server!(hyper::Response::builder() .version(hyper::Version::HTTP_2) - .body(hyper::Body::from("Hello World!")) + .body(http_body_util::Full::new(hyper::body::Bytes::from("Hello World!"))) .unwrap()); let id = api.request_start("POST", &format!("http://{}", addr)).unwrap();