[pbs-devel] [PATCH proxmox 13/17] proxmox-router: update to hyper 1.0

[Fabian Grünbichler]

and switch to proxmox_http's body. hyper now has a special (opaque) Body
implementation called Incoming which is used for incoming request bodies
on the server side, and incoming response bodies on the client side, so
our API handler's now consume an instance of this type.

Signed-off-by: Fabian Grünbichler <f.gruenbichler at proxmox.com>
---
 proxmox-router/Cargo.toml            |  6 ++++--
 proxmox-router/src/router.rs         | 19 +++++++++++--------
 proxmox-router/src/stream/parsing.rs | 16 +++++++++++-----
 3 files changed, 26 insertions(+), 15 deletions(-)

diff --git a/proxmox-router/Cargo.toml b/proxmox-router/Cargo.toml
index 9f1af3e9..de3f09e3 100644
--- a/proxmox-router/Cargo.toml
+++ b/proxmox-router/Cargo.toml
@@ -19,6 +19,7 @@ required-features = [ "cli" ]
 
 [dependencies]
 anyhow.workspace = true
+bytes = { workspace = true, optional = true }
 env_logger = { workspace = true, optional = true }
 futures.workspace = true
 http = { workspace = true, optional = true }
@@ -34,6 +35,7 @@ unicode-width ="0.1.8"
 rustyline = { version = "9", optional = true }
 libc = { workspace = true, optional = true }
 
+proxmox-http = { workspace = true, optional = true }
 proxmox-http-error.workspace = true
 proxmox-schema.workspace = true
 proxmox-async.workspace = true
@@ -45,6 +47,6 @@ tokio-stream.workspace = true
 [features]
 default = [ "cli", "server" ]
 cli = [ "stream", "dep:env_logger", "dep:libc", "dep:rustyline" ]
-server = [ "dep:http", "dep:hyper" ]
+server = [ "dep:http", "dep:hyper", "dep:proxmox-http", "proxmox-http?/body" ]
 test-harness = [ "proxmox-schema/test-harness" ]
-stream = [ "dep:hyper" ]
+stream = [ "dep:bytes", "dep:hyper", "dep:proxmox-http", "proxmox-http?/body" ]
diff --git a/proxmox-router/src/router.rs b/proxmox-router/src/router.rs
index 49593508..04552a91 100644
--- a/proxmox-router/src/router.rs
+++ b/proxmox-router/src/router.rs
@@ -8,9 +8,10 @@ use anyhow::Error;
 use http::request::Parts;
 #[cfg(feature = "server")]
 use http::{Method, Response};
-#[cfg(feature = "server")]
-use hyper::Body;
+use hyper::body::Incoming;
 use percent_encoding::percent_decode_str;
+#[cfg(feature = "server")]
+use proxmox_http::Body;
 use serde::Serialize;
 use serde_json::Value;
 
@@ -393,14 +394,15 @@ impl IntoRecord for Result<Record, Error> {
 /// ```
 /// # use serde_json::{json, Value};
 /// #
-/// use hyper::{Body, Response, http::request::Parts};
+/// use hyper::{Response, body::Incoming, http::request::Parts};
 ///
+/// use proxmox_http::Body;
 /// use proxmox_router::{ApiHandler, ApiMethod, ApiResponseFuture, RpcEnvironment};
 /// use proxmox_schema::ObjectSchema;
 ///
 /// fn low_level_hello(
 ///    parts: Parts,
-///    req_body: Body,
+///    req_body: Incoming,
 ///    param: Value,
 ///    info: &ApiMethod,
 ///    rpcenv: Box<dyn RpcEnvironment>,
@@ -408,7 +410,7 @@ impl IntoRecord for Result<Record, Error> {
 ///    Box::pin(async move {
 ///        let response = http::Response::builder()
 ///            .status(200)
-///            .body(Body::from("Hello world!"))?;
+///            .body(Body::from("Hello world!".to_string()))?;
 ///        Ok(response)
 ///    })
 /// }
@@ -421,7 +423,7 @@ impl IntoRecord for Result<Record, Error> {
 #[cfg(feature = "server")]
 pub type ApiAsyncHttpHandlerFn = &'static (dyn Fn(
     Parts,
-    Body,
+    Incoming,
     Value,
     &'static ApiMethod,
     Box<dyn RpcEnvironment>,
@@ -443,8 +445,9 @@ pub type ApiResponseFuture =
 /// ```
 /// use serde_json::Value;
 ///
-/// use hyper::{Body, Response, http::request::Parts};
+/// use hyper::{Response, http::request::Parts};
 ///
+/// use proxmox_http::Body;
 /// use proxmox_router::{ApiHandler, ApiMethod, ApiResponseFuture, RpcEnvironment};
 /// use proxmox_schema::ObjectSchema;
 ///
@@ -457,7 +460,7 @@ pub type ApiResponseFuture =
 ///    Box::pin(async move {
 ///        let response = http::Response::builder()
 ///            .status(200)
-///            .body(Body::from("Hello world!"))?;
+///            .body(Body::from("Hello world!".to_string()))?;
 ///        Ok(response)
 ///    })
 /// }
diff --git a/proxmox-router/src/stream/parsing.rs b/proxmox-router/src/stream/parsing.rs
index 1726d2f8..d9d00e59 100644
--- a/proxmox-router/src/stream/parsing.rs
+++ b/proxmox-router/src/stream/parsing.rs
@@ -4,10 +4,12 @@ use std::pin::Pin;
 use std::task::{ready, Poll};
 
 use anyhow::{format_err, Context as _, Error};
+use bytes::Bytes;
 use futures::io::{AsyncBufRead, AsyncBufReadExt, AsyncRead, BufReader};
-use hyper::body::{Body, Bytes};
 use serde::Deserialize;
 
+use proxmox_http::Body;
+
 use super::Record;
 
 pub struct Records<R = BodyBufReader>
@@ -269,8 +271,7 @@ impl AsyncBufRead for BodyBufReader {
         self: Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> Poll<io::Result<&[u8]>> {
-        use hyper::body::HttpBody;
-
+        use hyper::body::Body as HyperBody;
         let Self {
             ref mut reader,
             ref mut buf_at,
@@ -283,12 +284,17 @@ impl AsyncBufRead for BodyBufReader {
 
             let result = match reader {
                 None => return Poll::Ready(Ok(&[])),
-                Some(reader) => ready!(Pin::new(reader).poll_data(cx)),
+                Some(reader) => ready!(Pin::new(reader).poll_frame(cx)),
             };
 
             match result {
                 Some(Ok(bytes)) => {
-                    *buf_at = Some((bytes, 0));
+                    *buf_at = Some((
+                        bytes
+                            .into_data()
+                            .map_err(|_frame| io::Error::other("Failed to read frame from body"))?,
+                        0,
+                    ));
                 }
                 Some(Err(err)) => {
                     *reader = None;
-- 
2.39.5