Merge pull request #678 from swimos/deps

Upgrades hyper ecosystem

Author: SirCipher

Cargo.toml

@@ -115,7 +115,7 @@ tempdir = "0.3.7"
 slab = "0.4"
 smallvec = "1.10"
 uuid = "1.2"
-http = "0.2"
+http = "1.1.0"
 nom = "7.1"
 nom_locate = "4.0"
 tracing = "0.1"
@@ -132,8 +132,8 @@ proc-macro2 = "1.0"
 syn = "1.0"
 quote = "1.0.3"
 num-bigint = "0.4"
-ratchet = { package = "ratchet_rs", version = "0.4" }
-ratchet_fixture = "0.4"
+ratchet = { package = "ratchet_rs", version = "1.0" }
+ratchet_fixture = "1.0"
 flate2 = "1.0.22"
 bitflags = "2.5"
 rocksdb = "0.22"
@@ -147,7 +147,7 @@ trust-dns-resolver = "0.23.2"
 clap = "4.1"
 crossbeam-queue = { version = "0.3" }
 crossbeam-channel = { version = "0.5" }
-hyper = "0.14"
+hyper = "1.3.1"
 percent-encoding = "2.1.0"
 mime = "0.3"
 serde_json = "1.0"
@@ -166,9 +166,11 @@ quick-xml = "0.34.0"
 csv = "1.2"
 serde-xml-rs = "0.6"
 axum = "0.7.5"
-hyper-staticfile = "0.9"
+hyper-staticfile = "0.10.0"
 httparse = "1.8"
 sha-1 = "0.10.1"
 waker-fn = "1.1.0"
 num = "0.4"
 smol_str = "0.2.0"
+http-body-util = "0.1.2"
+hyper-util = "0.1.5"

example_apps/tutorial_app/Cargo.toml

@@ -10,7 +10,8 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "signal"] }
 example-util = { path = "../example_util" }
 rand = { workspace = true }
 tutorial-app-model = { path = "./model" }
-hyper = { workspace = true, features = ["server", "tcp", "http1"] }
+hyper = { workspace = true, features = ["server", "http1"] }
 hyper-staticfile = { workspace = true }
 http = { workspace = true }
-futures = { workspace = true }
+futures = { workspace = true }
+hyper-util = { workspace = true, features = ["server-graceful", "http1"] }

example_apps/tutorial_app/src/ui.rs

@@ -15,15 +15,16 @@
 use std::fmt::Display;
 use std::future::Future;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
-use std::pin::pin;
 use std::time::Duration;
 
-use futures::future::Either;
-use hyper::service::{make_service_fn, service_fn};
-use hyper::Server;
-use std::convert::Infallible;
-use std::path::Path;
+use futures::pin_mut;
+use hyper::server::conn::http1::Builder;
+use hyper::service::service_fn;
+use hyper_util::rt::TokioIo;
+use hyper_util::server::graceful::GracefulShutdown;
+
 use tokio::net::TcpListener;
+use tokio::select;
 use tokio::sync::oneshot;
 
 #[derive(Clone, Copy, Debug)]
@@ -47,42 +48,42 @@ where
     addr.await?;
     let bind_to = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0));
     let listener = TcpListener::bind(bind_to).await?;
-    let root = Path::new("static-files/");
-
+    let resolver = hyper_staticfile::Resolver::new("static-files/");
     let addr = listener.local_addr()?;
     println!("Web server bound to: {}", addr);
 
-    let make_svc = make_service_fn(move |_conn| async move {
-        Ok::<_, Infallible>(service_fn(move |request| async move {
-            let result = hyper_staticfile::resolve(&root, &request)
+    let make_svc = service_fn(move |request| {
+        let resolver = resolver.clone();
+        async move {
+            let result = resolver
+                .resolve_request(&request)
                 .await
                 .expect("Failed to access files.");
             let response = hyper_staticfile::ResponseBuilder::new()
                 .request(&request)
                 .build(result)?;
 
             Ok::<_, http::Error>(response)
-        }))
+        }
     });
 
-    let (stop_tx, stop_rx) = oneshot::channel();
-    let server = Server::from_tcp(listener.into_std()?)?
-        .serve(make_svc)
-        .with_graceful_shutdown(async move {
-            let _ = stop_rx.await;
-        });
+    let shutdown_handle = GracefulShutdown::new();
 
-    let shutdown = pin!(shutdown);
-    let server = pin!(server);
+    pin_mut!(shutdown, listener);
 
-    match futures::future::select(shutdown, server).await {
-        Either::Left((_, server)) => {
-            let _ = stop_tx.send(());
-            tokio::time::timeout(Duration::from_secs(2), server)
-                .await
-                .map_err(|_| TimeoutError)??;
-        }
-        Either::Right((result, _)) => result?,
+    loop {
+        let (stream, _) = select! {
+            biased;
+            _ = &mut shutdown => {
+                tokio::time::timeout(Duration::from_secs(2), shutdown_handle.shutdown())
+                   .await
+                   .map_err(|_| TimeoutError)?;
+               return Ok(());
+            }
+            result = listener.accept() => result?,
+        };
+
+        let conn = Builder::new().serve_connection(TokioIo::new(stream), make_svc.clone());
+        tokio::spawn(shutdown_handle.watch(conn));
     }
-    Ok(())
 }

runtime/swimos_http/Cargo.toml

@@ -19,7 +19,9 @@ sha-1 = { workspace = true }
 base64 = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+http-body-util = { workspace = true }
+hyper-util = { workspace = true, features = ["server-graceful", "http1", "server-auto"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["rt", "macros", "time"] }
-hyper = { workspace = true, features = ["server", "tcp", "http1"] }
+hyper = { workspace = true, features = ["server", "http1"] }

runtime/swimos_http/src/websocket.rs

@@ -1,22 +1,25 @@
+use std::{
+    collections::HashSet,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
 use base64::{engine::general_purpose::STANDARD, Engine};
 use bytes::{Bytes, BytesMut};
 use futures::{ready, Future, FutureExt};
 use http::{header::HeaderName, HeaderMap, HeaderValue, Method};
+use http_body_util::Full;
 use httparse::Header;
+use hyper::body::Incoming;
 use hyper::{
     upgrade::{OnUpgrade, Upgraded},
-    Body, Request, Response,
+    Request, Response,
 };
+use hyper_util::rt::TokioIo;
 use ratchet::{
     Extension, ExtensionProvider, NegotiatedExtension, Role, WebSocket, WebSocketConfig,
-    WebSocketStream,
 };
 use sha1::{Digest, Sha1};
-use std::{
-    collections::HashSet,
-    pin::Pin,
-    task::{Context, Poll},
-};
 use thiserror::Error;
 
 const UPGRADE_STR: &str = "Upgrade";
@@ -91,10 +94,12 @@ where
 }
 
 /// Produce a bad request response for a bad websocket upgrade request.
-pub fn fail_upgrade<ExtErr: std::error::Error>(error: UpgradeError<ExtErr>) -> Response<Body> {
+pub fn fail_upgrade<ExtErr: std::error::Error>(
+    error: UpgradeError<ExtErr>,
+) -> Response<Full<Bytes>> {
     Response::builder()
         .status(http::StatusCode::BAD_REQUEST)
-        .body(Body::from(error.to_string()))
+        .body(Full::from(error.to_string()))
         .expect(FAILED_RESPONSE)
 }
 
@@ -107,13 +112,12 @@ pub fn fail_upgrade<ExtErr: std::error::Error>(error: UpgradeError<ExtErr>) -> R
 /// * `unwrap_fn` - Used to unwrap the underlying socket type from the opaque [`Upgraded`] socket
 /// provided by hyper.
 pub fn upgrade<Ext, U>(
-    request: Request<Body>,
+    request: Request<Incoming>,
     negotiated: Negotiated<'_, Ext>,
     config: Option<WebSocketConfig>,
     unwrap_fn: U,
-) -> (Response<Body>, UpgradeFuture<Ext, U>)
+) -> (Response<Full<Bytes>>, UpgradeFuture<Ext, U>)
 where
-    U: SockUnwrap,
     Ext: Extension + Send,
 {
     let Negotiated {
@@ -149,7 +153,7 @@ where
         unwrap_fn,
     };
 
-    let response = builder.body(Body::empty()).expect(FAILED_RESPONSE);
+    let response = builder.body(Full::default()).expect(FAILED_RESPONSE);
     (response, fut)
 }
 
@@ -211,7 +215,7 @@ impl<ExtErr: std::error::Error> From<ExtErr> for UpgradeError<ExtErr> {
 /// The caller will generally know the real underlying type and this allows for that type to be
 /// restored.
 pub trait SockUnwrap {
-    type Sock: WebSocketStream;
+    type Sock;
 
     /// Unwrap the socket (returning the underlying socket and a buffer containing any bytes
     /// that have already been read).
@@ -222,15 +226,15 @@ pub trait SockUnwrap {
 pub struct NoUnwrap;
 
 impl SockUnwrap for NoUnwrap {
-    type Sock = Upgraded;
+    type Sock = TokioIo<Upgraded>;
 
     fn unwrap_sock(&self, upgraded: Upgraded) -> (Self::Sock, BytesMut) {
-        (upgraded, BytesMut::new())
+        (TokioIo::new(upgraded), BytesMut::new())
     }
 }
 
 /// A future that performs a websocket upgrade, unwraps the upgraded socket and
-/// creates a ratchet websocket from form it.
+/// creates a ratchet websocket from it.
 #[derive(Debug)]
 pub struct UpgradeFuture<Ext, U> {
     upgrade: OnUpgrade,

runtime/swimos_http/tests/wsserver.rs

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use bytes::BytesMut;
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{upgrade::Upgraded, Body, Request, Response, Server};
+use bytes::{Bytes, BytesMut};
+use hyper::service::service_fn;
+use hyper::{upgrade::Upgraded, Request, Response};
 use std::{
     error::Error,
-    net::{Ipv4Addr, SocketAddr, TcpListener},
+    net::{Ipv4Addr, SocketAddr},
     pin::pin,
     sync::Arc,
     time::Duration,
@@ -29,57 +29,63 @@ use futures::{
     future::{join, select, Either},
     Future,
 };
+use http_body_util::Full;
+use hyper::body::Incoming;
+use hyper_util::rt::{TokioExecutor, TokioIo};
+use hyper_util::server::conn::auto::Builder;
+use hyper_util::server::graceful::GracefulShutdown;
 use ratchet::{CloseCode, CloseReason, Message, NoExt, NoExtProvider, PayloadType, WebSocket};
 use thiserror::Error;
+use tokio::net::TcpListener;
 use tokio::{net::TcpSocket, sync::Notify};
 
 async fn run_server(
     bound_to: oneshot::Sender<SocketAddr>,
     done: Arc<Notify>,
 ) -> Result<(), Box<dyn Error>> {
-    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?;
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
     let bound = listener.local_addr()?;
     let _ = bound_to.send(bound);
 
-    let service = make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(upgrade_server)) });
-
-    let shutdown = Arc::new(Notify::new());
-    let shutdown_cpy = shutdown.clone();
-
-    let server = pin!(Server::from_tcp(listener)?
-        .serve(service)
-        .with_graceful_shutdown(async move {
-            shutdown_cpy.notified().await;
-        }));
+    let (io, _) = listener.accept().await?;
+    let builder = Builder::new(TokioExecutor::new());
+    let connection =
+        builder.serve_connection_with_upgrades(TokioIo::new(io), service_fn(upgrade_server));
+    let shutdown = GracefulShutdown::new();
 
+    let server = pin!(shutdown.watch(connection));
     let stop = pin!(done.notified());
+
     match select(server, stop).await {
-        Either::Left((result, _)) => result?,
-        Either::Right((_, server)) => {
-            shutdown.notify_one();
-            tokio::time::timeout(Duration::from_secs(2), server).await??;
+        Either::Left((result, _)) => match result {
+            Ok(()) => Ok(()),
+            Err(e) => Err(e),
+        },
+        Either::Right((_, _server)) => {
+            tokio::time::timeout(Duration::from_secs(2), shutdown.shutdown()).await?;
+            Ok(())
         }
     }
-
-    Ok(())
 }
 
-async fn upgrade_server(request: Request<Body>) -> Result<Response<Body>, hyper::http::Error> {
+async fn upgrade_server(
+    request: Request<Incoming>,
+) -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     let protocols = ["warp0"].into_iter().collect();
     match swimos_http::negotiate_upgrade(&request, &protocols, &NoExtProvider) {
         Ok(Some(negotiated)) => {
             let (response, upgraded) = swimos_http::upgrade(request, negotiated, None, NoUnwrap);
             tokio::spawn(run_websocket(upgraded));
             Ok(response)
         }
-        Ok(None) => Response::builder().body(Body::from("Success")),
+        Ok(None) => Response::builder().body(Full::from("Success")),
         Err(err) => Ok(swimos_http::fail_upgrade(err)),
     }
 }
 
 async fn run_websocket<F>(upgrade_fut: F)
 where
-    F: Future<Output = Result<WebSocket<Upgraded, NoExt>, hyper::Error>> + Send,
+    F: Future<Output = Result<WebSocket<TokioIo<Upgraded>, NoExt>, hyper::Error>> + Send,
 {
     match upgrade_fut.await {
         Ok(mut websocket) => {

runtime/swimos_remote/Cargo.toml

@@ -36,6 +36,7 @@ either = { workspace = true }
 smallvec = { workspace = true }
 url = { workspace = true }
 pin-project = { workspace = true }
+hyper = { workspace = true }
 
 rustls = { workspace = true, optional = true }
 webpki = { workspace = true, optional = true }

runtime/swimos_remote/src/net/mod.rs

@@ -103,6 +103,7 @@ where
 /// Provides all networking functionality required for a Warp client (DNS resolution and opening sockets).
 pub trait ClientConnections: Clone + Send + Sync + 'static {
     type ClientSocket: AsyncRead + AsyncWrite + Unpin + Send + Sync + 'static;
+
     fn try_open(
         &self,
         scheme: Scheme,
@@ -111,6 +112,7 @@ pub trait ClientConnections: Clone + Send + Sync + 'static {
     ) -> BoxFuture<'_, ConnectionResult<Self::ClientSocket>>;
 
     fn dns_resolver(&self) -> BoxDnsResolver;
+
     fn lookup(
         &self,
         host: String,

server/swimos_server_app/Cargo.toml

@@ -39,10 +39,12 @@ rand = { workspace = true }
 url = { workspace = true }
 swimos_rocks_store = { workspace = true, optional = true }
 parking_lot = { workspace = true }
-hyper = { workspace = true, features = ["server", "runtime", "tcp", "http1", "backports"] }
+hyper = { workspace = true, features = ["server", "http1"] }
 pin-project = { workspace = true }
 percent-encoding = { workspace = true }
 rustls = { workspace = true }
+http-body-util = { workspace = true }
+hyper-util = { workspace = true }
 
 [dev-dependencies]
 swimos_recon = { workspace = true }