Update to http 1.0 and hyper 1.1 on hard mode

crates/twirp/Cargo.toml

@@ -22,8 +22,15 @@ reqwest = { version = "0.11", features = ["default", "gzip", "json"], optional =
 url = { version = "2.5", optional = true }
 
 # For the server feature
-hyper = { version = "0.14", features = ["full"], optional = true }
+bytes = "1.5"
+http-body-util = "0.1"
+hyper = { version = "1.1", features = ["full"], optional = true }
+hyper-util = { version = "0.1", features = ["tokio"] }
+pin-project-lite = "0.2"
 
 # For the test-support feature
 async-trait = { version = "0.1", optional = true }
 tokio = { version = "1.33", features = [], optional = true }
+
+[dev-dependencies]
+tokio = { version = "1.33", features = ["rt", "macros"] }

crates/twirp/src/body.rs

@@ -0,0 +1,112 @@
+use std::fmt::{self, Debug, Formatter};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use bytes::Bytes;
+use http_body_util::combinators::UnsyncBoxBody;
+use http_body_util::BodyExt;
+use hyper::body::Frame;
+
+use crate::GenericError;
+
+type BoxBody = UnsyncBoxBody<Bytes, GenericError>;
+
+pin_project_lite::pin_project! {
+    /// Generic body type (like `axum::body::Body`).
+    pub struct Body {
+        #[pin]
+        inner: BoxBody
+    }
+}
+
+impl Debug for Body {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Body").finish()
+    }
+}
+
+impl From<Bytes> for Body {
+    fn from(bytes: Bytes) -> Self {
+        Body {
+            inner: BoxBody::new(http_body_util::Full::new(bytes).map_err(|err| match err {})),
+        }
+    }
+}
+
+impl From<Vec<u8>> for Body {
+    fn from(bytes: Vec<u8>) -> Self {
+        Bytes::from(bytes).into()
+    }
+}
+
+impl From<String> for Body {
+    fn from(text: String) -> Self {
+        Bytes::from(text).into()
+    }
+}
+
+impl From<&'static str> for Body {
+    fn from(text: &'static str) -> Self {
+        Bytes::from(text).into()
+    }
+}
+
+impl Body {
+    /// Create a new Body that wraps another `http::body::Body`.
+    pub fn new<B>(body: B) -> Self
+    where
+        B: hyper::body::Body<Data = Bytes> + Send + 'static,
+        B::Error: Into<GenericError>,
+    {
+        Body {
+            inner: BoxBody::new(body.map_err(|err| err.into())),
+        }
+    }
+
+    pub fn empty() -> Self {
+        Self::new(http_body_util::Empty::new())
+    }
+
+    pub(crate) fn protobuf<T>(message: &T) -> Self
+    where
+        T: prost::Message,
+    {
+        serialize_proto_message(message).into()
+    }
+
+    pub(crate) fn json<T>(data: &T) -> Result<Self, serde_json::Error>
+    where
+        T: serde::Serialize,
+    {
+        let json = serde_json::to_string(&data)?;
+        Ok(json.into())
+    }
+}
+
+pub(crate) fn serialize_proto_message<T>(m: &T) -> Vec<u8>
+where
+    T: prost::Message,
+{
+    let len = m.encoded_len();
+    let mut data = Vec::with_capacity(len);
+    m.encode(&mut data)
+        .expect("can only fail if buffer does not have capacity");
+    assert_eq!(data.len(), len);
+    data
+}
+
+impl hyper::body::Body for Body {
+    /// Values yielded by the `Body`.
+    type Data = bytes::Bytes;
+
+    /// The error type this `Body` might generate.
+    type Error = GenericError;
+
+    /// Attempt to pull out the next data buffer of this stream.
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        self.project().inner.poll_frame(cx)
+    }
+}

crates/twirp/src/client.rs

@@ -1,13 +1,13 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use hyper::header::{InvalidHeaderValue, CONTENT_TYPE};
-use hyper::StatusCode;
+use reqwest::header::{InvalidHeaderValue, CONTENT_TYPE};
+use reqwest::StatusCode;
 use thiserror::Error;
 use url::Url;
 
 use crate::headers::{CONTENT_TYPE_JSON, CONTENT_TYPE_PROTOBUF};
-use crate::{to_proto_body, TwirpErrorResponse};
+use crate::{body, TwirpErrorResponse};
 
 #[derive(Debug, Error)]
 pub enum ClientError {
@@ -146,7 +146,7 @@ impl Client {
             .http_client
             .post(url)
             .header(CONTENT_TYPE, CONTENT_TYPE_PROTOBUF)
-            .body(to_proto_body(body))
+            .body(body::serialize_proto_message(&body))
             .build()?;
 
         // Create and execute the middleware handlers

crates/twirp/src/error.rs

@@ -2,11 +2,13 @@
 
 use std::collections::HashMap;
 
-use hyper::{header, Body, Response, StatusCode};
+use hyper::{header, Response, StatusCode};
 use serde::{Deserialize, Serialize, Serializer};
 
+use crate::Body;
+
 // Alias for a generic error
-pub type GenericError = Box<dyn std::error::Error + Send + Sync>;
+pub type GenericError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 macro_rules! twirp_error_codes {
     (

crates/twirp/src/lib.rs

@@ -1,13 +1,15 @@
 #[cfg(feature = "client")]
 pub mod client;
 
+mod body;
 pub mod error;
 pub mod headers;
 pub mod server;
 
 #[cfg(any(test, feature = "test-support"))]
 pub mod test;
 
+pub use body::Body;
 pub use client::{Client, ClientBuilder, ClientError, Middleware, Next, Result};
 pub use error::*; // many constructors like `invalid_argument()`
 pub use server::{serve, Router, Timings};
@@ -17,15 +19,3 @@ pub use reqwest;
 
 // Re-export `url so that the generated code works without additional dependencies beyond just the `twirp` crate.
 pub use url;
-
-pub(crate) fn to_proto_body<T>(m: T) -> hyper::Body
-where
-    T: prost::Message,
-{
-    let len = m.encoded_len();
-    let mut data = Vec::with_capacity(len);
-    m.encode(&mut data)
-        .expect("can only fail if buffer does not have capacity");
-    assert_eq!(data.len(), len);
-    hyper::Body::from(data)
-}