fix http upgrade with new hyper version

1 files changed, 17 insertions, 23 deletions

diff --git a/src/filters/proxy.rs b/src/filters/proxy.rs
index 11ff8c1..ad959ad 100644
--- a/src/filters/proxy.rs
+++ b/src/filters/proxy.rs
@@ -1,13 +1,7 @@
 use crate::{helper::TokioIo, ServiceError, State};
 use http_body_util::{combinators::BoxBody, BodyExt};
-use hyper::{
-    body::Incoming,
-    header::UPGRADE,
-    http::{uri::PathAndQuery, HeaderValue},
-    upgrade::OnUpgrade,
-    Request, Uri,
-};
-use log::{debug, error, warn};
+use hyper::{body::Incoming, http::HeaderValue, upgrade::OnUpgrade, Request, StatusCode};
+use log::{debug, warn};
 use std::{net::SocketAddr, sync::Arc};
 use tokio::net::TcpStream;
 
@@ -20,23 +14,23 @@ pub async fn proxy_request(
     #[cfg(feature = "mond")]
     state.reporting.request_out.inc();
 
-    *req.uri_mut() = Uri::builder()
-        .path_and_query(
-            req.uri()
-                .clone()
-                .path_and_query()
-                .cloned()
-                .unwrap_or(PathAndQuery::from_static("/")),
-        )
-        .build()
-        .unwrap();
+    //? Do we know what this did?
+    // *req.uri_mut() = Uri::builder()
+    //     .path_and_query(
+    //         req.uri()
+    //             .clone()
+    //             .path_and_query()
+    //             .cloned()
+    //             .unwrap_or(PathAndQuery::from_static("/")),
+    //     )
+    //     .build()
+    //     .unwrap();
 
     req.headers_mut().insert(
         "x-real-ip",
         HeaderValue::from_str(&format!("{}", addr.ip())).unwrap(),
     );
 
-    let do_upgrade = req.headers().contains_key(UPGRADE);
     let on_upgrade_downstream = req.extensions_mut().remove::<OnUpgrade>();
 
     let _limit_guard = state.l_outgoing.try_acquire()?;
@@ -52,7 +46,7 @@ pub async fn proxy_request(
             .await
             .map_err(ServiceError::Hyper)?;
         tokio::task::spawn(async move {
-            if let Err(err) = conn.await {
+            if let Err(err) = conn.with_upgrades().await {
                 warn!("connection failed: {:?}", err);
             }
         });
@@ -62,14 +56,14 @@ pub async fn proxy_request(
             .map_err(ServiceError::Hyper)?
     };
 
-    if do_upgrade {
+    if resp.status() == StatusCode::SWITCHING_PROTOCOLS {
         let on_upgrade_upstream = resp
             .extensions_mut()
             .remove::<OnUpgrade>()
             .ok_or(ServiceError::UpgradeFailed)?;
         let on_upgrade_downstream = on_upgrade_downstream.ok_or(ServiceError::UpgradeFailed)?;
         tokio::task::spawn(async move {
-            debug!("about to upgrade connection, sending empty response");
+            debug!("about to upgrade connection");
             match (on_upgrade_upstream.await, on_upgrade_downstream.await) {
                 (Ok(upgraded_upstream), Ok(upgraded_downstream)) => {
                     debug!("upgrade successful");
@@ -85,7 +79,7 @@ pub async fn proxy_request(
                         Err(e) => warn!("proxy socket error: {e}"),
                     }
                 }
-                (a, b) => error!("upgrade error: upstream={a:?} downstream={b:?}"),
+                (a, b) => warn!("upgrade error: upstream={a:?} downstream={b:?}"),
             }
         });
     }