Merge pull request #24 from blazzy/send-stream-response

Ensure stream responses are Send
blazzy · Aug 23, 2024 · a50ee12 · a50ee12
2 parents 9400be1 + dc593eb
commit a50ee12
Show file tree

Hide file tree

Showing 11 changed files with 45 additions and 12 deletions.
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -39,6 +39,7 @@ jobs:
 
     - name: Build Podman
       run: |
+        sudo apt-get update
         sudo apt-get install \
           btrfs-progs \
           golang-go \

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Changelog
 
+## Unreleased
+
+* `Send` constraint added to byte stream responses ([#24](https://github.com/blazzy/podman-rest-client/pull/24))
+
 ## v0.12.2 - 2024-08-21
 
 * Improve formatting in generated docs with extra whitespace and removal redundant lines

diff --git a/openapi-client-gen/src/generate/rust_hyper_legacy/templates/apis.rs b/openapi-client-gen/src/generate/rust_hyper_legacy/templates/apis.rs
@@ -49,7 +49,7 @@ pub fn operations(
                 .unwrap_or_else(|| Ok(quote! { () }))?;
 
             let return_type = if Some(true) == success.map(|m| m.data.is_stream()) {
-                quote! { Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> }
+                quote! { Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> }
             } else {
                 quote! { Pin<Box<dyn Future<Output=Result<#response, Error>> + Send + 'a>> }
             };

diff --git a/openapi-client-gen/src/generate/rust_hyper_legacy/templates/request.rs b/openapi-client-gen/src/generate/rust_hyper_legacy/templates/request.rs
@@ -49,7 +49,7 @@ pub async fn execute_request_bytes(
 pub fn execute_request_stream<'a>(
     config: &'a dyn ClientConfig,
     request: Result<http::request::Request<String>, Error>,
-) -> Pin<Box<dyn Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+) -> Pin<Box<dyn Stream<Item = Result<bytes::Bytes, Error>> + 'a + Send>> {
     let result = async move {
         let response = config.request(request?).await?;
         let status = response.status();
@@ -73,6 +73,6 @@ pub fn execute_request_stream<'a>(
     Box::pin(stream::once(result).flat_map(|result| match result {
         Ok(stream) => stream,
         Err(err) => Box::pin(stream::once(async { Err(err) }))
-            as Pin<Box<dyn Stream<Item = Result<_, Error>>>>,
+            as Pin<Box<dyn Stream<Item = Result<_, Error>> + Send>>,
     }))
 }
diff --git a/src/api_common/request.rs b/src/api_common/request.rs
@@ -49,7 +49,7 @@ pub async fn execute_request_bytes(
 pub fn execute_request_stream<'a>(
     config: &'a dyn ClientConfig,
     request: Result<http::request::Request<String>, Error>,
-) -> Pin<Box<dyn Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+) -> Pin<Box<dyn Stream<Item = Result<bytes::Bytes, Error>> + 'a + Send>> {
     let result = async move {
         let response = config.request(request?).await?;
         let status = response.status();
@@ -73,6 +73,6 @@ pub fn execute_request_stream<'a>(
     Box::pin(stream::once(result).flat_map(|result| match result {
         Ok(stream) => stream,
         Err(err) => Box::pin(stream::once(async { Err(err) }))
-            as Pin<Box<dyn Stream<Item = Result<_, Error>>>>,
+            as Pin<Box<dyn Stream<Item = Result<_, Error>> + Send>>,
     }))
 }
diff --git a/src/v4/apis/images.rs b/src/v4/apis/images.rs
@@ -243,7 +243,7 @@ pub trait Images: HasConfig + Send + Sync {
         &'a self,
         name: &'a str,
         params: Option<crate::v4::params::ImageGetLibpod<'a>>,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {

diff --git a/src/v4/apis/images_compat.rs b/src/v4/apis/images_compat.rs
@@ -161,7 +161,7 @@ pub trait ImagesCompat: HasConfig + Send + Sync {
     fn image_get<'a>(
         &'a self,
         name: &'a str,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {

diff --git a/src/v5/apis/containers.rs b/src/v5/apis/containers.rs
@@ -422,7 +422,7 @@ pub trait Containers: HasConfig + Send + Sync {
     fn container_export_libpod<'a>(
         &'a self,
         name: &'a str,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {

diff --git a/src/v5/apis/images.rs b/src/v5/apis/images.rs
@@ -249,7 +249,7 @@ pub trait Images: HasConfig + Send + Sync {
         &'a self,
         name: &'a str,
         params: Option<crate::v5::params::ImageGetLibpod<'a>>,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {
@@ -526,7 +526,7 @@ pub trait Images: HasConfig + Send + Sync {
     fn image_export_libpod<'a>(
         &'a self,
         params: Option<crate::v5::params::ImageExportLibpod<'a>>,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {

diff --git a/src/v5/apis/images_compat.rs b/src/v5/apis/images_compat.rs
@@ -167,7 +167,7 @@ pub trait ImagesCompat: HasConfig + Send + Sync {
     fn image_get<'a>(
         &'a self,
         name: &'a str,
-    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + 'a>> {
+    ) -> Pin<Box<dyn futures::stream::Stream<Item = Result<bytes::Bytes, Error>> + Send + 'a>> {
         request::execute_request_stream(
             self.get_config(),
             (|| {

diff --git a/tests/v5.rs b/tests/v5.rs
@@ -51,7 +51,7 @@ async fn it_can_create_a_volume() {
 }
 
 #[tokio::test]
-async fn it_can_run_in_a_thread() {
+async fn it_can_run_call_async_methods_in_threads() {
     common::test_init().await;
 
     let config = Config::guess().await.unwrap();
@@ -65,6 +65,34 @@ async fn it_can_run_in_a_thread() {
     assert!(handle.await.is_ok());
 }
 
+#[tokio::test]
+async fn it_can_call_stream_functions_in_a_thread() {
+    common::test_init().await;
+
+    let config = Config::guess().await.unwrap();
+    let client = PodmanRestClient::new(config).await.unwrap();
+
+    common::pull_nginx_image(&client).await;
+    common::create_nginx_container(&client, "podman_rest_client_stream_thread_test").await;
+
+    let config = Config::guess().await.unwrap();
+    let client = PodmanRestClient::new(config).await.unwrap();
+
+    let handle = tokio::spawn(async move {
+        let stream = client
+            .containers()
+            .container_export_libpod("podman_rest_client_stream_thread_test");
+        let bytes: Vec<u8> = stream.map_ok(|b| b.to_vec()).try_concat().await.unwrap();
+        let c = std::io::Cursor::new(bytes);
+
+        assert!(tar::Archive::new(c).entries().unwrap().count() > 0);
+
+        common::delete_container(&client, "podman_rest_client_stream_thread_test").await;
+    });
+
+    assert!(handle.await.is_ok());
+}
+
 #[tokio::test]
 async fn it_can_pull_images() {
     common::test_init().await;