Merge pull request #66 from PocketRelay/62-task-upgrading-axum-to-v07…

…-and-hyper-to-v1 feat: upgrade hyper and request
PocketRelay · Apr 13, 2024 · 33606c2 · 33606c2
2 parents 319bac5 + e7a0d83
commit 33606c2
Show file tree

Hide file tree

Showing 8 changed files with 492 additions and 455 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -47,7 +47,8 @@ email_address = "0.2.4"
 tokio-util = { version = "0.7", features = ["codec"] }
 
 # Hyper for connection upgrades (TODO: Update to 1.0 once reqwest supports it)
-hyper = "0.14.25"
+hyper = "1.2.0"
+hyper-util = { version = "0.1.3", features = ["tokio"] }
 tower = "0.4"
 
 bitflags = { version = "2.3.1", features = ["serde"] }
@@ -97,13 +98,13 @@ features = [
 
 # Axum web framework (TODO: Update to 0.7 once once reqwest supports hyper 1.0)
 [dependencies.axum]
-version = "0.6.1"
+version = "0.7.5"
 default-features = false
 features = ["http1", "json", "query", "tokio"]
 
 # HTTP Client
 [dependencies.reqwest]
-version = "0.11.12"
+version = "0.12.3"
 default-features = false
 features = ["json", "rustls-tls"]
 

diff --git a/src/main.rs b/src/main.rs
@@ -7,11 +7,11 @@ use crate::{
     },
     utils::signing::SigningKey,
 };
-use axum::{Extension, Server};
+use axum::{self, Extension};
 use config::load_config;
 use log::{debug, error, info, LevelFilter};
 use std::{net::SocketAddr, sync::Arc};
-use tokio::{join, signal};
+use tokio::{join, net::TcpListener, signal};
 use utils::logging;
 
 mod config;
@@ -88,13 +88,22 @@ async fn main() {
 
     info!("Starting server on {} (v{})", addr, VERSION);
 
-    if let Err(err) = Server::bind(&addr)
-        .serve(router)
+    // Start the TCP listener
+    let listener = match TcpListener::bind(addr).await {
+        Ok(value) => value,
+        Err(err) => {
+            error!("Failed to bind HTTP server pm {}: {:?}", addr, err);
+            return;
+        }
+    };
+
+    // Run the HTTP server
+    if let Err(err) = axum::serve(listener, router)
         .with_graceful_shutdown(async move {
             _ = signal::ctrl_c().await;
         })
         .await
     {
-        error!("Failed to bind HTTP server on {}: {:?}", addr, err);
+        error!("Error within HTTP server {:?}", err);
     }
 }
diff --git a/src/middleware/auth.rs b/src/middleware/auth.rs
@@ -6,7 +6,7 @@ use crate::{
     services::sessions::{Sessions, VerifyError},
 };
 use axum::{
-    body::boxed,
+    body::Body,
     extract::FromRequestParts,
     http::StatusCode,
     response::{IntoResponse, Response},
@@ -125,6 +125,6 @@ impl IntoResponse for TokenError {
             Self::Database(_) => StatusCode::INTERNAL_SERVER_ERROR,
         };
 
-        (status, boxed(self.to_string())).into_response()
+        (status, Body::from(self.to_string())).into_response()
     }
 }
diff --git a/src/middleware/cors.rs b/src/middleware/cors.rs
@@ -1,15 +1,17 @@
 use axum::{
-    http::{header, HeaderValue, Method, Request, StatusCode},
+    body::Body,
+    http::{header, HeaderValue, Method, StatusCode},
     middleware::Next,
     response::Response,
 };
+use hyper::Request;
 
 /// Middleware layer function for appending CORS headers to requests
 /// and responding to options requests
 ///
 /// `req`  The request to handle
 /// `next` The next layer to use
-pub async fn cors_layer<T>(req: Request<T>, next: Next<T>) -> Response {
+pub async fn cors_layer(req: Request<Body>, next: Next) -> Response {
     // Create a new response for OPTIONS requests
     let mut res: Response = if req.method() == Method::OPTIONS {
         // Default response for OPTIONS requests
@@ -42,12 +44,12 @@ pub async fn cors_layer<T>(req: Request<T>, next: Next<T>) -> Response {
 #[cfg(test)]
 mod test {
     use super::cors_layer;
-    use axum::{middleware::from_fn, routing::get, Router};
+    use axum::{body::Body, middleware::from_fn, routing::get, Router};
     use hyper::{
         header::{
             ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN,
         },
-        Body, Method, Request, StatusCode,
+        Method, Request, StatusCode,
     };
     use tower::ServiceExt;
 

diff --git a/src/routes/public.rs b/src/routes/public.rs
@@ -1,5 +1,5 @@
 use axum::{
-    body::Full,
+    body::Body,
     http::{HeaderValue, Request},
     response::{IntoResponse, Response},
 };
@@ -35,7 +35,7 @@ fn find_local_path(path: &str) -> Option<PathBuf> {
     Some(file_path)
 }
 
-impl<T> Service<Request<T>> for PublicContent {
+impl Service<Request<Body>> for PublicContent {
     type Response = Response;
     type Error = Infallible;
     type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
@@ -44,7 +44,7 @@ impl<T> Service<Request<T>> for PublicContent {
         Poll::Ready(Ok(()))
     }
 
-    fn call(&mut self, req: Request<T>) -> Self::Future {
+    fn call(&mut self, req: Request<Body>) -> Self::Future {
         let path = req.uri().path();
 
         // Strip the leading slash in order to match paths correctly
@@ -86,7 +86,7 @@ impl<T> Service<Request<T>> for PublicContent {
                 if local_path.exists() && local_path.is_file() {
                     if let Ok(contents) = tokio::fs::read(local_path).await {
                         // Create byte reponse from the embedded file
-                        let mut response = Full::from(contents).into_response();
+                        let mut response = Body::from(contents).into_response();
                         response
                             .headers_mut()
                             .insert(CONTENT_TYPE, HeaderValue::from_static(mime_type));
@@ -98,7 +98,7 @@ impl<T> Service<Request<T>> for PublicContent {
             // File exists within binary serve that
             if let Some(contents) = Self::get(&path) {
                 // Create byte response from the embedded file
-                let mut response = Full::from(contents).into_response();
+                let mut response = Body::from(contents).into_response();
                 response
                     .headers_mut()
                     .insert(CONTENT_TYPE, HeaderValue::from_static(mime_type));

diff --git a/src/services/tunnel.rs b/src/services/tunnel.rs
@@ -6,6 +6,7 @@ use self::codec::{TunnelCodec, TunnelMessage};
 use crate::utils::{hashing::IntHashMap, types::GameID};
 use futures_util::{Sink, Stream};
 use hyper::upgrade::Upgraded;
+use hyper_util::rt::TokioIo;
 use parking_lot::RwLock;
 use std::{
     collections::HashMap,
@@ -232,7 +233,7 @@ pub struct Tunnel {
     id: TunnelId,
     /// The IO tunnel used to send information to the host and receive
     /// response
-    io: Framed<Upgraded, TunnelCodec>,
+    io: Framed<TokioIo<Upgraded>, TunnelCodec>,
     /// Receiver for messages that should be written to the tunnel
     rx: mpsc::UnboundedReceiver<TunnelMessage>,
     /// Future state for writing to the `io`
@@ -282,7 +283,7 @@ impl Tunnel {
         let (tx, rx) = mpsc::unbounded_channel();
 
         // Wrap the `io` with the [`TunnelCodec`] for framing
-        let io = Framed::new(io, TunnelCodec::default());
+        let io = Framed::new(TokioIo::new(io), TunnelCodec::default());
 
         // Acquire the tunnel ID
         let id = service.next_tunnel_id.fetch_add(1, Ordering::AcqRel);

diff --git a/src/session/mod.rs b/src/session/mod.rs
@@ -29,6 +29,7 @@ use crate::{
 };
 use futures_util::{future::BoxFuture, Sink, Stream};
 use hyper::upgrade::Upgraded;
+use hyper_util::rt::TokioIo;
 use log::{debug, log_enabled, warn};
 use parking_lot::Mutex;
 use serde::Serialize;
@@ -239,7 +240,7 @@ impl Session {
         });
 
         SessionFuture {
-            io: Framed::new(io, PacketCodec::default()),
+            io: Framed::new(TokioIo::new(io), PacketCodec::default()),
             router: &router,
             rx,
             session: session.clone(),
@@ -482,7 +483,7 @@ impl Debug for DebugSessionData {
 /// Future for processing a session
 struct SessionFuture<'a> {
     /// The IO for reading and writing
-    io: Framed<Upgraded, PacketCodec>,
+    io: Framed<TokioIo<Upgraded>, PacketCodec>,
     /// Receiver for packets to write
     rx: mpsc::UnboundedReceiver<Packet>,
     /// The session this link is for