Merge pull request #99 from paritytech/chore-update-hyper

chore(deps): update hyper v1.0
paritytech · Mar 7, 2024 · 3222405 · 3222405
2 parents afe56f5 + 420216a
commit 3222405
Show file tree

Hide file tree

Showing 5 changed files with 50 additions and 20 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -27,17 +27,18 @@ httparse = { default-features = false, features = ["std"], version = "1.3.4" }
 log = { default-features = false, version = "0.4.8" }
 rand = { default-features = false, features = ["std", "std_rng"], version = "0.8" }
 sha1 = { default-features = false, version = "0.10" }
-http = { default-features = false, version = "0.2", optional = true }
+http = { version = "1", optional = true }
 
 [dev-dependencies]
 quickcheck = "1"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7", features = ["compat"] }
 tokio-stream = { version = "0.1", features = ["net"] }
-hyper = { version = "0.14.10", features = ["full"] }
+http-body-util = "0.1"
+hyper = { version = "1.2", features = ["full"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
 env_logger = "0.11.1"
 
 [[example]]
 name = "hyper_server"
 required-features = ["http"]
-
diff --git a/examples/hyper_server.rs b/examples/hyper_server.rs
@@ -21,33 +21,62 @@
 //
 // You'll see any messages you send echoed back.
 
+use std::net::SocketAddr;
+
 use futures::io::{BufReader, BufWriter};
-use hyper::{Body, Request, Response};
+use hyper::server::conn::http1;
+use hyper::{body::Bytes, service::service_fn, Request, Response};
+use hyper_util::rt::TokioIo;
 use soketto::{
 	handshake::http::{is_upgrade_request, Server},
 	BoxedError,
 };
 use tokio_util::compat::TokioAsyncReadCompatExt;
 
+type FullBody = http_body_util::Full<Bytes>;
+
 /// Start up a hyper server.
 #[tokio::main]
 async fn main() -> Result<(), BoxedError> {
 	env_logger::init();
 
-	let addr = ([127, 0, 0, 1], 3000).into();
+	let addr: SocketAddr = ([127, 0, 0, 1], 3000).into();
+	let listener = tokio::net::TcpListener::bind(addr).await?;
 
-	let service =
-		hyper::service::make_service_fn(|_| async { Ok::<_, hyper::Error>(hyper::service::service_fn(handler)) });
-	let server = hyper::Server::bind(&addr).serve(service);
+	log::info!(
+		"Listening on http://{:?} — connect and I'll echo back anything you send!",
+		listener.local_addr().unwrap()
+	);
 
-	println!("Listening on http://{} — connect and I'll echo back anything you send!", server.local_addr());
-	server.await?;
+	loop {
+		let stream = match listener.accept().await {
+			Ok((stream, addr)) => {
+				log::info!("Accepting new connection: {addr}");
+				stream
+			}
+			Err(e) => {
+				log::error!("Accepting new connection failed: {e}");
+				continue;
+			}
+		};
 
-	Ok(())
+		tokio::spawn(async {
+			let io = TokioIo::new(stream);
+			let conn = http1::Builder::new().serve_connection(io, service_fn(handler));
+
+			// Enable upgrades on the connection for the websocket upgrades to work.
+			let conn = conn.with_upgrades();
+
+			// Log any errors that might have occurred during the connection.
+			if let Err(err) = conn.await {
+				log::error!("HTTP connection failed {err}");
+			}
+		});
+	}
 }
 
 /// Handle incoming HTTP Requests.
-async fn handler(req: Request<Body>) -> Result<hyper::Response<Body>, BoxedError> {
+async fn handler(req: Request<hyper::body::Incoming>) -> Result<hyper::Response<FullBody>, BoxedError> {
 	if is_upgrade_request(&req) {
 		// Create a new handshake server.
 		let mut server = Server::new();
@@ -69,30 +98,31 @@ async fn handler(req: Request<Body>) -> Result<hyper::Response<Body>, BoxedError
 						log::error!("Error upgrading to websocket connection: {}", e);
 					}
 				});
-				Ok(response.map(|()| Body::empty()))
+				Ok(response.map(|()| FullBody::default()))
 			}
 			// We tried to upgrade and failed early on; tell the client about the failure however we like:
 			Err(e) => {
 				log::error!("Could not upgrade connection: {}", e);
-				Ok(Response::new(Body::from("Something went wrong upgrading!")))
+				Ok(Response::new(FullBody::from("Something went wrong upgrading!")))
 			}
 		}
 	} else {
 		// The request wasn't an upgrade request; let's treat it as a standard HTTP request:
-		Ok(Response::new(Body::from("Hello HTTP!")))
+		Ok(Response::new(FullBody::from("Hello HTTP!")))
 	}
 }
 
 /// Echo any messages we get from the client back to them
-async fn websocket_echo_messages(server: Server, req: Request<Body>) -> Result<(), BoxedError> {
+async fn websocket_echo_messages(server: Server, req: Request<hyper::body::Incoming>) -> Result<(), BoxedError> {
 	// The negotiation to upgrade to a WebSocket connection has been successful so far. Next, we get back the underlying
 	// stream using `hyper::upgrade::on`, and hand this to a Soketto server to use to handle the WebSocket communication
 	// on this socket.
 	//
 	// Note: awaiting this won't succeed until the handshake response has been returned to the client, so this must be
 	// spawned on a separate task so as not to block that response being handed back.
 	let stream = hyper::upgrade::on(req).await?;
-	let stream = BufReader::new(BufWriter::new(stream.compat()));
+	let io = TokioIo::new(stream);
+	let stream = BufReader::new(BufWriter::new(io.compat()));
 
 	// Get back a reader and writer that we can use to send and receive websocket messages.
 	let (mut sender, mut receiver) = server.into_builder(stream).finish();

diff --git a/src/base.rs b/src/base.rs
@@ -18,7 +18,7 @@
 //! [base]: https://tools.ietf.org/html/rfc6455#section-5.2
 
 use crate::{as_u64, Parsing};
-use std::{convert::TryFrom, fmt, io};
+use std::{fmt, io};
 
 /// Max. size of a frame header.
 pub(crate) const MAX_HEADER_SIZE: usize = 14;

diff --git a/src/data.rs b/src/data.rs
@@ -8,7 +8,7 @@
 
 //! Types describing various forms of payload data.
 
-use std::{convert::TryFrom, fmt};
+use std::fmt;
 
 use crate::connection::CloseReason;
 

diff --git a/src/handshake/http.rs b/src/handshake/http.rs
@@ -21,7 +21,6 @@ use crate::handshake;
 use bytes::BytesMut;
 use futures::prelude::*;
 use http::{header, HeaderMap, Response};
-use std::convert::TryInto;
 use std::mem;
 
 /// A re-export of [`handshake::Error`].