+
+fn make_send<S>(mut stream: S) -> impl Stream<Item = S::Item> + Send + Unpin + 'static
+where
+ S: Stream + Unpin + 'static,
+ S::Item: Send,
+{
+ // NOTE: the 8 here is arbitrary
+ let (tx, rx) = tokio::sync::mpsc::channel(8);
+
+ // NOTE: spawning stream into a new task can potentially hit this bug:
+ // - https://github.com/actix/actix-web/issues/1679
+ //
+ // Since 4.0.0-beta.2 this issue is incredibly less frequent. I have not personally reproduced it.
+ // That said, it is still technically possible to encounter.
+ actix_web::rt::spawn(async move {
+ while let Some(res) = stream.next().await {
+ if tx.send(res).await.is_err() {
+ break;
+ }
+ }
+ });
+
+ SendStream { rx }
+}
+
+struct SendStream<T> {
+ rx: tokio::sync::mpsc::Receiver<T>,
+}
+
+impl<T> Stream for SendStream<T>
+where
+ T: Send,
+{
+ type Item = T;
+
+ fn poll_next(
+ mut self: std::pin::Pin<&mut Self>,
+ cx: &mut std::task::Context<'_>,
+ ) -> std::task::Poll<Option<Self::Item>> {
+ std::pin::Pin::new(&mut self.rx).poll_recv(cx)
+ }
+}