chore: upgrade async-channel

Cargo.toml

@@ -61,7 +61,7 @@ resolver = "2"
 [workspace.dependencies]
 adaptive_backoff = "0.2.1"
 anyhow = "1.0.86"
-async-channel = { version = "1.9.0", default-features = false }
+async-channel = { version = "2.3.1", default-features = false }
 async-io = "2.3.3"
 async-lock = "3.4.0"
 async-std = { version = "1.8.0", default-features = false }
@@ -192,7 +192,7 @@ fluvio-smartmodule = { version = "0.8.0", path = "crates/fluvio-smartmodule", de
 fluvio-socket = { version = "0.15.1", path = "crates/fluvio-socket", default-features = false }
 fluvio-spu-schema = { version = "0.17.0", path = "crates/fluvio-spu-schema", default-features  = false }
 fluvio-storage = { path = "crates/fluvio-storage" }
-fluvio-stream-dispatcher = { version = "0.13.2", path = "crates/fluvio-stream-dispatcher" }
+fluvio-stream-dispatcher = { version = "0.13.3", path = "crates/fluvio-stream-dispatcher" }
 fluvio-stream-model = { version = "0.11.4", path = "crates/fluvio-stream-model", default-features = false }
 fluvio-types = { version = "0.5.0", path = "crates/fluvio-types", default-features = false }
 fluvio-kv-storage = { path = "crates/fluvio-kv-storage", default-features = false }

crates/fluvio-cli/src/client/consume/mod.rs

@@ -394,7 +394,7 @@ mod cmd {
             let mut stream = fluvio
                 .consumer_with_config(consume_config)
                 .await?
-                .take_until(stop_signal.recv());
+                .take_until(stop_signal.recv().boxed());
             self.consume_records_stream(&mut stream, tableformat)
                 .await?;
 

crates/fluvio-sc/src/controllers/mirroring/controller.rs

@@ -71,7 +71,7 @@ impl<C: MetadataItem> RemoteMirrorController<C> {
             if let Some((home, _)) = self.get_mirror_home_cluster().await {
                 info!(home = %home.id,  "connected to home cluster");
                 let home_config = self.build_home_client(&home).await?;
-                let mut stream = self.request_stream(&home, home_config).await?;
+                let mut stream = self.request_stream(&home, home_config).await?.boxed();
 
                 info!("created request stream");
 

crates/fluvio-stream-dispatcher/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-stream-dispatcher"
 edition = "2021"
-version = "0.13.7"
+version = "0.13.8"
 authors = ["Fluvio Contributors <[email protected]>"]
 description = "Fluvio Event Stream access"
 repository = "https://github.com/infinyon/fluvio"

crates/fluvio/src/admin.rs

@@ -349,21 +349,23 @@ impl FluvioAdmin {
         debug!(api_version = req_msg.header.api_version(), obj = %S::LABEL, "create watch stream");
         let inner_socket = self.socket.new_socket();
         let stream = inner_socket.create_stream(req_msg, 10).await?;
-        Ok(stream.map(|respons_result| match respons_result {
-            Ok(response) => {
-                let watch_response = response.downcast().map_err(|err| {
-                    IoError::new(ErrorKind::Other, format!("downcast error: {:#?}", err))
-                })?;
-                watch_response.ok_or(IoError::new(
+        Ok(stream
+            .map(|respons_result| match respons_result {
+                Ok(response) => {
+                    let watch_response = response.downcast().map_err(|err| {
+                        IoError::new(ErrorKind::Other, format!("downcast error: {:#?}", err))
+                    })?;
+                    watch_response.ok_or(IoError::new(
+                        ErrorKind::Other,
+                        format!("cannot decoded as {s}", s = S::LABEL),
+                    ))
+                }
+                Err(err) => Err(IoError::new(
                     ErrorKind::Other,
-                    format!("cannot decoded as {s}", s = S::LABEL),
-                ))
-            }
-            Err(err) => Err(IoError::new(
-                ErrorKind::Other,
-                format!("socket error {err}"),
-            )),
-        }))
+                    format!("socket error {err}"),
+                )),
+            })
+            .boxed())
     }
 }
 

crates/fluvio/src/consumer/mod.rs

@@ -384,10 +384,11 @@ where
             warn!("SPU does not support Offset Management API");
         }
 
-        let mut stream = self
-            .pool
-            .create_stream_with_version(&replica, stream_request, stream_fetch_version)
-            .await?;
+        let mut stream = StreamExt::boxed(
+            self.pool
+                .create_stream_with_version(&replica, stream_request, stream_fetch_version)
+                .await?,
+        );
 
         let (server_sender, server_recv) =
             async_channel::bounded::<StreamToServer>(STREAM_TO_SERVER_CHANNEL_SIZE);

crates/fluvio/src/consumer/stream.rs

@@ -110,7 +110,8 @@ impl<T: Stream<Item = Result<Record, ErrorCode>> + Unpin> Stream
     }
 }
 
-impl<T> ConsumerStream for futures_util::stream::TakeUntil<T, async_channel::Recv<'_, ()>>
+impl<T> ConsumerStream
+    for futures_util::stream::TakeUntil<T, BoxFuture<'_, async_channel::Recv<'_, ()>>>
 where
     T: ConsumerStream,
 {

crates/fluvio/src/sync/controller.rs

@@ -76,16 +76,18 @@ where
     }
 
     #[instrument(
-        skip(self,response),
+        skip(self,async_response),
         fields(
             spec = S::LABEL,
         )
     )]
-    async fn dispatch_loop(mut self, mut response: AsyncResponse<ObjectApiWatchRequest>) {
+    async fn dispatch_loop(mut self, async_response: AsyncResponse<ObjectApiWatchRequest>) {
         use tokio::select;
 
         debug!("{} starting dispatch loop", S::LABEL);
 
+        let mut response = async_response.boxed();
+
         loop {
             // check if shutdown is set
             if self.shutdown.is_set() {

crates/fluvio/src/sync/mod.rs

@@ -168,7 +168,7 @@ mod context {
     mod unstable {
         use super::*;
         use crate::metadata::store::MetadataChanges;
-        use futures_util::Stream;
+        use futures_util::{Stream, StreamExt};
 
         impl<S> StoreContext<S>
         where
@@ -193,7 +193,7 @@ mod context {
                     }
                 });
 
-                receiver
+                receiver.boxed()
             }
         }
     }