Merge pull request #1291 from GeorgeHahn/hyper-update

Update hyper to 1.x
librespot-org · Jun 9, 2024 · 1504430 · 1504430
2 parents 8f9bec2 + bd5c284
commit 1504430
Show file tree

Hide file tree

Showing 12 changed files with 281 additions and 334 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/audio/Cargo.toml b/audio/Cargo.toml
@@ -19,7 +19,9 @@ bytes = "1"
 ctr = "0.9"
 futures-core = "0.3"
 futures-util = "0.3"
-hyper = { version = "0.14", features = ["client"] }
+hyper = { version = "1.3", features = [] }
+hyper-util = { version = "0.1", features = ["client"] }
+http-body-util = "0.1.1"
 log = "0.4"
 parking_lot = { version = "0.12", features = ["deadlock_detection"] }
 tempfile = "3"

diff --git a/audio/src/fetch/mod.rs b/audio/src/fetch/mod.rs
@@ -12,7 +12,8 @@ use std::{
 };
 
 use futures_util::{future::IntoStream, StreamExt, TryFutureExt};
-use hyper::{client::ResponseFuture, header::CONTENT_RANGE, Body, Response, StatusCode};
+use hyper::{body::Incoming, header::CONTENT_RANGE, Response, StatusCode};
+use hyper_util::client::legacy::ResponseFuture;
 use parking_lot::{Condvar, Mutex};
 use tempfile::NamedTempFile;
 use thiserror::Error;
@@ -133,7 +134,7 @@ pub enum AudioFile {
 #[derive(Debug)]
 pub struct StreamingRequest {
     streamer: IntoStream<ResponseFuture>,
-    initial_response: Option<Response<Body>>,
+    initial_response: Option<Response<Incoming>>,
     offset: usize,
     length: usize,
 }

diff --git a/audio/src/fetch/receive.rs b/audio/src/fetch/receive.rs
@@ -7,6 +7,7 @@ use std::{
 
 use bytes::Bytes;
 use futures_util::StreamExt;
+use http_body_util::BodyExt;
 use hyper::StatusCode;
 use tempfile::NamedTempFile;
 use tokio::sync::{mpsc, oneshot};
@@ -97,7 +98,7 @@ async fn receive_data(
         }
 
         let body = response.into_body();
-        let data = match hyper::body::to_bytes(body).await {
+        let data = match body.collect().await.map(|b| b.to_bytes()) {
             Ok(bytes) => bytes,
             Err(e) => break Err(e.into()),
         };

diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -25,10 +25,12 @@ futures-util = { version = "0.3", features = ["alloc", "bilock", "sink", "unstab
 governor = { version = "0.6", default-features = false, features = ["std", "jitter"] }
 hmac = "0.12"
 httparse = "1.7"
-http = "0.2"
-hyper = { version = "0.14", features = ["client", "http1", "http2", "tcp"] }
-hyper-proxy = { version = "0.9", default-features = false, features = ["rustls"] }
-hyper-rustls = { version = "0.24", features = ["http2"] }
+http = "1.0"
+hyper = { version = "1.3", features = ["http1", "http2"] }
+hyper-util = { version = "0.1", features = ["client"] }
+http-body-util = "0.1.1"
+hyper-proxy2 = { version = "0.1", default-features = false, features = ["rustls"] }
+hyper-rustls = { version = "0.26", features = ["http2"] }
 log = "0.4"
 nonzero_ext = "0.3"
 num-bigint = { version = "0.4", features = ["rand"] }

diff --git a/core/src/apresolve.rs b/core/src/apresolve.rs
@@ -1,6 +1,7 @@
 use std::collections::VecDeque;
 
-use hyper::{Body, Method, Request};
+use bytes::Bytes;
+use hyper::{Method, Request};
 use serde::Deserialize;
 
 use crate::Error;
@@ -85,7 +86,7 @@ impl ApResolver {
         let req = Request::builder()
             .method(Method::GET)
             .uri("https://apresolve.spotify.com/?type=accesspoint&type=dealer&type=spclient")
-            .body(Body::empty())?;
+            .body(Bytes::new())?;
 
         let body = self.session().http_client().request_body(req).await?;
         let data: ApResolveData = serde_json::from_slice(body.as_ref())?;

diff --git a/core/src/error.rs b/core/src/error.rs
@@ -321,18 +321,14 @@ impl From<http::Error> for Error {
 
 impl From<hyper::Error> for Error {
     fn from(err: hyper::Error) -> Self {
-        if err.is_parse() || err.is_parse_too_large() || err.is_parse_status() || err.is_user() {
+        if err.is_parse() || err.is_parse_status() || err.is_user() {
             return Self::new(ErrorKind::Internal, err);
         }
 
         if err.is_canceled() {
             return Self::new(ErrorKind::Cancelled, err);
         }
 
-        if err.is_connect() {
-            return Self::new(ErrorKind::Unavailable, err);
-        }
-
         if err.is_incomplete_message() {
             return Self::new(ErrorKind::DataLoss, err);
         }
@@ -349,6 +345,16 @@ impl From<hyper::Error> for Error {
     }
 }
 
+impl From<hyper_util::client::legacy::Error> for Error {
+    fn from(err: hyper_util::client::legacy::Error) -> Self {
+        if err.is_connect() {
+            return Self::new(ErrorKind::Unavailable, err);
+        }
+
+        Self::new(ErrorKind::Unknown, err)
+    }
+}
+
 impl From<time::error::Parse> for Error {
     fn from(err: time::error::Parse) -> Self {
         Self::new(ErrorKind::FailedPrecondition, err)

diff --git a/core/src/http_client.rs b/core/src/http_client.rs
@@ -10,13 +10,14 @@ use governor::{
     clock::MonotonicClock, middleware::NoOpMiddleware, state::InMemoryState, Quota, RateLimiter,
 };
 use http::{header::HeaderValue, Uri};
-use hyper::{
-    client::{HttpConnector, ResponseFuture},
-    header::USER_AGENT,
-    Body, Client, HeaderMap, Request, Response, StatusCode,
-};
-use hyper_proxy::{Intercept, Proxy, ProxyConnector};
+use http_body_util::{BodyExt, Full};
+use hyper::{body::Incoming, header::USER_AGENT, HeaderMap, Request, Response, StatusCode};
+use hyper_proxy2::{Intercept, Proxy, ProxyConnector};
 use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
+use hyper_util::{
+    client::legacy::{connect::HttpConnector, Client, ResponseFuture},
+    rt::TokioExecutor,
+};
 use nonzero_ext::nonzero;
 use once_cell::sync::OnceCell;
 use parking_lot::Mutex;
@@ -89,7 +90,7 @@ impl From<HttpClientError> for Error {
     }
 }
 
-type HyperClient = Client<ProxyConnector<HttpsConnector<HttpConnector>>, Body>;
+type HyperClient = Client<ProxyConnector<HttpsConnector<HttpConnector>>, Full<bytes::Bytes>>;
 
 pub struct HttpClient {
     user_agent: HeaderValue,
@@ -146,7 +147,7 @@ impl HttpClient {
     fn try_create_hyper_client(proxy_url: Option<&Url>) -> Result<HyperClient, Error> {
         // configuring TLS is expensive and should be done once per process
         let https_connector = HttpsConnectorBuilder::new()
-            .with_native_roots()
+            .with_native_roots()?
             .https_or_http()
             .enable_http1()
             .enable_http2()
@@ -160,7 +161,7 @@ impl HttpClient {
         };
         let proxy_connector = ProxyConnector::from_proxy(https_connector, proxy)?;
 
-        let client = Client::builder()
+        let client = Client::builder(TokioExecutor::new())
             .http2_adaptive_window(true)
             .build(proxy_connector);
         Ok(client)
@@ -171,23 +172,20 @@ impl HttpClient {
             .get_or_try_init(|| Self::try_create_hyper_client(self.proxy_url.as_ref()))
     }
 
-    pub async fn request(&self, req: Request<Body>) -> Result<Response<Body>, Error> {
+    pub async fn request(&self, req: Request<Bytes>) -> Result<Response<Incoming>, Error> {
         debug!("Requesting {}", req.uri().to_string());
 
         // `Request` does not implement `Clone` because its `Body` may be a single-shot stream.
         // As correct as that may be technically, we now need all this boilerplate to clone it
         // ourselves, as any `Request` is moved in the loop.
-        let (parts, body) = req.into_parts();
-        let body_as_bytes = hyper::body::to_bytes(body)
-            .await
-            .unwrap_or_else(|_| Bytes::new());
+        let (parts, body_as_bytes) = req.into_parts();
 
         loop {
             let mut req = Request::builder()
                 .method(parts.method.clone())
                 .uri(parts.uri.clone())
                 .version(parts.version)
-                .body(Body::from(body_as_bytes.clone()))?;
+                .body(body_as_bytes.clone())?;
             *req.headers_mut() = parts.headers.clone();
 
             let request = self.request_fut(req)?;
@@ -212,20 +210,21 @@ impl HttpClient {
                 }
             }
 
-            return Ok(response?);
+            let response = response?;
+            return Ok(response);
         }
     }
 
-    pub async fn request_body(&self, req: Request<Body>) -> Result<Bytes, Error> {
+    pub async fn request_body(&self, req: Request<Bytes>) -> Result<Bytes, Error> {
         let response = self.request(req).await?;
-        Ok(hyper::body::to_bytes(response.into_body()).await?)
+        Ok(response.into_body().collect().await?.to_bytes())
     }
 
-    pub fn request_stream(&self, req: Request<Body>) -> Result<IntoStream<ResponseFuture>, Error> {
+    pub fn request_stream(&self, req: Request<Bytes>) -> Result<IntoStream<ResponseFuture>, Error> {
         Ok(self.request_fut(req)?.into_stream())
     }
 
-    pub fn request_fut(&self, mut req: Request<Body>) -> Result<ResponseFuture, Error> {
+    pub fn request_fut(&self, mut req: Request<Bytes>) -> Result<ResponseFuture, Error> {
         let headers_mut = req.headers_mut();
         headers_mut.insert(USER_AGENT, self.user_agent.clone());
 
@@ -251,7 +250,7 @@ impl HttpClient {
             ))
         })?;
 
-        Ok(self.hyper_client()?.request(req))
+        Ok(self.hyper_client()?.request(req.map(Full::new)))
     }
 
     pub fn get_retry_after(headers: &HeaderMap<HeaderValue>) -> Option<Duration> {

diff --git a/core/src/spclient.rs b/core/src/spclient.rs
@@ -10,10 +10,10 @@ use data_encoding::HEXUPPER_PERMISSIVE;
 use futures_util::future::IntoStream;
 use http::header::HeaderValue;
 use hyper::{
-    client::ResponseFuture,
     header::{HeaderName, ACCEPT, AUTHORIZATION, CONTENT_TYPE, RANGE},
-    Body, HeaderMap, Method, Request,
+    HeaderMap, Method, Request,
 };
+use hyper_util::client::legacy::ResponseFuture;
 use protobuf::{Enum, Message, MessageFull};
 use rand::RngCore;
 use sha1::{Digest, Sha1};
@@ -156,7 +156,7 @@ impl SpClient {
             .method(&Method::POST)
             .uri("https://clienttoken.spotify.com/v1/clienttoken")
             .header(ACCEPT, HeaderValue::from_static("application/x-protobuf"))
-            .body(Body::from(body))?;
+            .body(body.into())?;
 
         self.session().http_client().request_body(request).await
     }
@@ -465,7 +465,7 @@ impl SpClient {
             let mut request = Request::builder()
                 .method(method)
                 .uri(url)
-                .body(Body::from(body.to_owned()))?;
+                .body(body.to_owned().into())?;
 
             // Reconnection logic: keep getting (cached) tokens because they might have expired.
             let token = self
@@ -727,7 +727,7 @@ impl SpClient {
                 RANGE,
                 HeaderValue::from_str(&format!("bytes={}-{}", offset, offset + length - 1))?,
             )
-            .body(Body::empty())?;
+            .body(Bytes::new())?;
 
         let stream = self.session().http_client().request_stream(req)?;
 
@@ -738,7 +738,7 @@ impl SpClient {
         let request = Request::builder()
             .method(&Method::GET)
             .uri(url)
-            .body(Body::empty())?;
+            .body(Bytes::new())?;
 
         self.session().http_client().request_body(request).await
     }

diff --git a/discovery/Cargo.toml b/discovery/Cargo.toml
@@ -11,14 +11,17 @@ edition = "2021"
 [dependencies]
 aes = "0.8"
 base64 = "0.21"
+bytes = "1"
 cfg-if = "1.0"
 ctr = "0.9"
 dns-sd = { version = "0.1.3", optional = true }
 form_urlencoded = "1.0"
 futures-core = "0.3"
 futures-util = "0.3"
 hmac = "0.12"
-hyper = { version = "0.14", features = ["http1", "server", "tcp"] }
+hyper = { version = "1.3", features = ["http1"] }
+hyper-util = { version = "0.1", features = ["server-auto", "server-graceful", "service"] }
+http-body-util = "0.1.1"
 libmdns = "0.8"
 log = "0.4"
 rand = "0.8"

diff --git a/discovery/src/lib.rs b/discovery/src/lib.rs
@@ -124,7 +124,7 @@ impl Builder {
     pub fn launch(self) -> Result<Discovery, Error> {
         let mut port = self.port;
         let name = self.server_config.name.clone().into_owned();
-        let server = DiscoveryServer::new(self.server_config, &mut port)??;
+        let server = DiscoveryServer::new(self.server_config, &mut port)?;
         let _zeroconf_ip = self.zeroconf_ip;
         let svc;