boxed the receiver

Author: sehz

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?.boxed_local();
+                let mut stream = self.request_stream(&home, home_config).await?.boxed();
 
                 info!("created request stream");
 

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 =  StreamExt::boxed(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/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()
             }
         }
     }