Hyper 1/Axum 0.7 support (#33)

* Support for Hyper v1/Axum 0.7 * Working text streamer * CSV stream fix * JSON support * Protobuf support * Fixed tests * Updated pipelines
abdolence · Dec 8, 2023 · 7eb9f69 · 7eb9f69
1 parent a81bf93
commit 7eb9f69
Show file tree

Hide file tree

Showing 14 changed files with 76 additions and 98 deletions.
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -3,12 +3,15 @@ on:
   workflow_dispatch:
   push:
   pull_request:
-    types: [opened]  
+    types: [opened]
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_protected && github.run_id || github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
 jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608 # v4
+      - uses: actions/checkout@v4
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal

diff --git a/Cargo.toml b/Cargo.toml
@@ -21,10 +21,11 @@ name = "axum_streams"
 path = "src/lib.rs"
 
 [dependencies]
-axum = { version = "0.6" }
+axum = { version = "0.7" }
 bytes = "1"
 futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
-http = "0.2"
+http = "1"
+http-body = "1"
 mime = "0.3"
 tokio = "1"
 serde = { version = "1", features = ["serde_derive"], optional = true }
@@ -44,10 +45,10 @@ text = []
 
 [dev-dependencies]
 futures = "0.3"
-hyper = "0.14"
+hyper = "1"
 reqwest = { version = "0.11", default-features = false, features = ["json", "stream", "multipart"] }
 tower = { version = "0.4", default-features = false, features = ["util", "make"] }
-tower-http = { version = "0.4", features = ["util", "map-response-body"] }
+tower-http = { version = "0.5", features = ["util", "map-response-body"] }
 tower-layer = "0.3"
 tower-service = "0.3"
 tokio = { version = "1", features = ["full"] }

diff --git a/README.md b/README.md
@@ -19,13 +19,14 @@ and want to avoid huge memory allocation.
 Cargo.toml:
 ```toml
 [dependencies]
-axum-streams = { version = "0.10", features=["json", "csv", "protobuf", "text"] }
+axum-streams = { version = "0.11", features=["json", "csv", "protobuf", "text"] }
 ```
 
 ## Compatibility matrix
 
 | axum | axum-streams |
 |------|--------------|
+| 0.7  | v0.11        |
 | 0.6  | v0.9-v0.10   |
 | 0.5  | 0.7          |
 

diff --git a/examples/csv-example.rs b/examples/csv-example.rs
@@ -1,28 +1,30 @@
 use axum::response::IntoResponse;
 use axum::routing::*;
 use axum::Router;
-use std::net::SocketAddr;
 
 use futures::prelude::*;
 use serde::{Deserialize, Serialize};
+use tokio::net::TcpListener;
 use tokio_stream::StreamExt;
 
 use axum_streams::*;
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 struct MyTestStructure {
-    some_test_field: String,
+    some_test_field1: String,
+    some_test_field2: String,
 }
 
 fn source_test_stream() -> impl Stream<Item = MyTestStructure> {
     // Simulating a stream with a plain vector and throttling to show how it works
     stream::iter(vec![
         MyTestStructure {
-            some_test_field: "test1".to_string()
+            some_test_field1: "test1".to_string(),
+            some_test_field2: "test2".to_string()
         };
         1000
     ])
-    .throttle(std::time::Duration::from_millis(50))
+    .throttle(std::time::Duration::from_millis(500))
 }
 
 async fn test_csv_stream() -> impl IntoResponse {
@@ -45,10 +47,7 @@ async fn main() {
         // `GET /` goes to `root`
         .route("/csv-stream", get(test_csv_stream));
 
-    let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
+    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
 
-    axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .await
-        .unwrap();
+    axum::serve(listener, app).await.unwrap();
 }
diff --git a/examples/json-example.rs b/examples/json-example.rs
@@ -1,10 +1,10 @@
 use axum::response::IntoResponse;
 use axum::routing::*;
 use axum::Router;
-use std::net::SocketAddr;
 
 use futures::prelude::*;
 use serde::{Deserialize, Serialize};
+use tokio::net::TcpListener;
 use tokio_stream::StreamExt;
 
 use axum_streams::*;
@@ -41,10 +41,7 @@ async fn main() {
         .route("/json-array-stream", get(test_json_array_stream))
         .route("/json-nl-stream", get(test_json_nl_stream));
 
-    let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
+    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
 
-    axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .await
-        .unwrap();
+    axum::serve(listener, app).await.unwrap();
 }
diff --git a/examples/protobuf-example.rs b/examples/protobuf-example.rs
@@ -1,9 +1,9 @@
 use axum::response::IntoResponse;
 use axum::routing::*;
 use axum::Router;
-use std::net::SocketAddr;
 
 use futures::prelude::*;
+use tokio::net::TcpListener;
 use tokio_stream::StreamExt;
 
 use axum_streams::*;
@@ -36,10 +36,6 @@ async fn main() {
         // `GET /` goes to `root`
         .route("/protobuf-stream", get(test_protobuf_stream));
 
-    let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
-
-    axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .await
-        .unwrap();
+    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
+    axum::serve(listener, app).await.unwrap();
 }
diff --git a/examples/text-example.rs b/examples/text-example.rs
@@ -1,9 +1,9 @@
 use axum::response::IntoResponse;
 use axum::routing::*;
 use axum::Router;
-use std::net::SocketAddr;
 
 use futures::prelude::*;
+use tokio::net::TcpListener;
 use tokio_stream::StreamExt;
 
 use axum_streams::*;
@@ -28,10 +28,7 @@ async fn main() {
         // `GET /` goes to `root`
         .route("/text-stream", get(test_text_stream));
 
-    let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
+    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
 
-    axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .await
-        .unwrap();
+    axum::serve(listener, app).await.unwrap();
 }
diff --git a/src/csv_format.rs b/src/csv_format.rs
@@ -3,6 +3,7 @@ use futures::Stream;
 use futures_util::stream::BoxStream;
 use futures_util::StreamExt;
 use http::HeaderMap;
+use http_body::Frame;
 use serde::Serialize;
 
 pub struct CsvStreamFormat {
@@ -96,7 +97,7 @@ where
     fn to_bytes_stream<'a, 'b>(
         &'a self,
         stream: BoxStream<'b, T>,
-    ) -> BoxStream<'b, Result<axum::body::Bytes, axum::Error>> {
+    ) -> BoxStream<'b, Result<Frame<axum::body::Bytes>, axum::Error>> {
         let stream_with_header = self.has_headers;
         let stream_delimiter = self.delimiter;
         let stream_flexible = self.flexible;
@@ -106,7 +107,7 @@ where
         let stream_escape = self.escape;
         let terminator = self.terminator;
 
-        let stream_bytes: BoxStream<Result<axum::body::Bytes, axum::Error>> = Box::pin({
+        let stream_bytes: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> = Box::pin({
             stream.enumerate().map(move |(index, obj)| {
                 let mut writer = csv::WriterBuilder::new()
                     .has_headers(index == 0 && stream_with_header)
@@ -125,6 +126,7 @@ where
                     .into_inner()
                     .map_err(axum::Error::new)
                     .map(axum::body::Bytes::from)
+                    .map(Frame::data)
             })
         });
 
@@ -188,7 +190,7 @@ mod tests {
             }),
         );
 
-        let client = TestClient::new(app);
+        let client = TestClient::new(app).await;
 
         let expected_csv = test_stream_vec
             .iter()

diff --git a/src/json_formats.rs b/src/json_formats.rs
@@ -4,6 +4,7 @@ use futures::Stream;
 use futures_util::stream::BoxStream;
 use futures_util::StreamExt;
 use http::HeaderMap;
+use http_body::Frame;
 use serde::Serialize;
 use std::io::Write;
 
@@ -22,8 +23,8 @@ where
     fn to_bytes_stream<'a, 'b>(
         &'a self,
         stream: BoxStream<'b, T>,
-    ) -> BoxStream<'b, Result<axum::body::Bytes, axum::Error>> {
-        let stream_bytes: BoxStream<Result<axum::body::Bytes, axum::Error>> = Box::pin({
+    ) -> BoxStream<'b, Result<Frame<axum::body::Bytes>, axum::Error>> {
+        let stream_bytes: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> = Box::pin({
             stream.enumerate().map(|(index, obj)| {
                 let mut buf = BytesMut::new().writer();
 
@@ -37,7 +38,7 @@ where
                 match sep_write_res {
                     Ok(_) => {
                         match serde_json::to_writer(&mut buf, &obj).map_err(axum::Error::new) {
-                            Ok(_) => Ok(buf.into_inner().freeze()),
+                            Ok(_) => Ok(Frame::data(buf.into_inner().freeze())),
                             Err(e) => Err(e),
                         }
                     }
@@ -46,14 +47,14 @@ where
             })
         });
 
-        let prepend_stream: BoxStream<Result<axum::body::Bytes, axum::Error>> =
+        let prepend_stream: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> =
             Box::pin(futures_util::stream::once(futures_util::future::ready(
-                Ok::<_, axum::Error>(axum::body::Bytes::from(JSON_ARRAY_BEGIN_BYTES)),
+                Ok::<_, axum::Error>(Frame::data(axum::body::Bytes::from(JSON_ARRAY_BEGIN_BYTES))),
             )));
 
-        let append_stream: BoxStream<Result<axum::body::Bytes, axum::Error>> =
+        let append_stream: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> =
             Box::pin(futures_util::stream::once(futures_util::future::ready(
-                Ok::<_, axum::Error>(axum::body::Bytes::from(JSON_ARRAY_END_BYTES)),
+                Ok::<_, axum::Error>(Frame::data(axum::body::Bytes::from(JSON_ARRAY_END_BYTES))),
             )));
 
         Box::pin(prepend_stream.chain(stream_bytes.chain(append_stream)))
@@ -84,13 +85,13 @@ where
     fn to_bytes_stream<'a, 'b>(
         &'a self,
         stream: BoxStream<'b, T>,
-    ) -> BoxStream<'b, Result<axum::body::Bytes, axum::Error>> {
-        let stream_bytes: BoxStream<Result<axum::body::Bytes, axum::Error>> = Box::pin({
+    ) -> BoxStream<'b, Result<Frame<axum::body::Bytes>, axum::Error>> {
+        let stream_bytes: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> = Box::pin({
             stream.map(|obj| {
                 let mut buf = BytesMut::new().writer();
                 match serde_json::to_writer(&mut buf, &obj).map_err(axum::Error::new) {
                     Ok(_) => match buf.write_all(JSON_NL_SEP_BYTES).map_err(axum::Error::new) {
-                        Ok(_) => Ok(buf.into_inner().freeze()),
+                        Ok(_) => Ok(Frame::data(buf.into_inner().freeze())),
                         Err(e) => Err(e),
                     },
                     Err(e) => Err(e),
@@ -164,7 +165,7 @@ mod tests {
             get(|| async { StreamBodyAs::new(JsonArrayStreamFormat::new(), test_stream) }),
         );
 
-        let client = TestClient::new(app);
+        let client = TestClient::new(app).await;
 
         let expected_json = serde_json::to_string(&test_stream_vec).unwrap();
         let res = client.get("/").send().await.unwrap();
@@ -201,7 +202,7 @@ mod tests {
             get(|| async { StreamBodyAs::new(JsonNewLineStreamFormat::new(), test_stream) }),
         );
 
-        let client = TestClient::new(app);
+        let client = TestClient::new(app).await;
 
         let expected_json = test_stream_vec
             .iter()

diff --git a/src/protobuf_format.rs b/src/protobuf_format.rs
@@ -3,6 +3,7 @@ use futures::Stream;
 use futures_util::stream::BoxStream;
 use futures_util::StreamExt;
 use http::HeaderMap;
+use http_body::Frame;
 
 pub struct ProtobufStreamFormat;
 
@@ -19,7 +20,7 @@ where
     fn to_bytes_stream<'a, 'b>(
         &'a self,
         stream: BoxStream<'b, T>,
-    ) -> BoxStream<'b, Result<axum::body::Bytes, axum::Error>> {
+    ) -> BoxStream<'b, Result<Frame<axum::body::Bytes>, axum::Error>> {
         fn write_protobuf_record<T>(obj: T) -> Result<Vec<u8>, axum::Error>
         where
             T: prost::Message,
@@ -33,10 +34,12 @@ where
             Ok(frame_vec)
         }
 
-        let stream_bytes: BoxStream<Result<axum::body::Bytes, axum::Error>> = Box::pin({
+        let stream_bytes: BoxStream<Result<Frame<axum::body::Bytes>, axum::Error>> = Box::pin({
             stream.map(move |obj| {
                 let write_protobuf_res = write_protobuf_record(obj);
-                write_protobuf_res.map(axum::body::Bytes::from)
+                write_protobuf_res
+                    .map(axum::body::Bytes::from)
+                    .map(Frame::data)
             })
         });
 
@@ -97,7 +100,7 @@ mod tests {
             get(|| async { StreamBodyAs::new(ProtobufStreamFormat::new(), test_stream) }),
         );
 
-        let client = TestClient::new(app);
+        let client = TestClient::new(app).await;
 
         let expected_proto_buf: Vec<u8> = test_stream_vec
             .iter()

diff --git a/src/stream_body_as.rs b/src/stream_body_as.rs
@@ -1,15 +1,16 @@
 use crate::stream_format::StreamingFormat;
-use axum::body::HttpBody;
+use axum::body::{Body, HttpBody};
 use axum::response::{IntoResponse, Response};
 use futures::Stream;
 use futures_util::stream::BoxStream;
 use http::HeaderMap;
+use http_body::Frame;
 use std::fmt::Formatter;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub struct StreamBodyAs<'a> {
-    stream: BoxStream<'a, Result<axum::body::Bytes, axum::Error>>,
+    stream: BoxStream<'a, Result<Frame<axum::body::Bytes>, axum::Error>>,
     trailers: Option<HeaderMap>,
 }
 
@@ -46,7 +47,7 @@ impl IntoResponse for StreamBodyAs<'static> {
             HeaderMap::new()
         };
 
-        let mut response = Response::new(axum::body::boxed(self));
+        let mut response: Response<Body> = Response::new(Body::new(self));
         *response.headers_mut() = headers;
         response
     }
@@ -56,17 +57,10 @@ impl<'a> HttpBody for StreamBodyAs<'a> {
     type Data = axum::body::Bytes;
     type Error = axum::Error;
 
-    fn poll_data(
+    fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         Pin::new(&mut self.stream).poll_next(cx)
     }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
-        Poll::Ready(Ok(self.trailers.clone()))
-    }
 }