StreamExt::scan: specified lifetime requirements

futures-util/src/stream/stream/mod.rs

@@ -566,10 +566,10 @@ pub trait StreamExt: Stream {
     /// assert_eq!(vec![1, 2, 3], stream.collect::<Vec<_>>().await);
     /// # });
     /// ```
-    fn scan<S, B, Fut, F>(self, initial_state: S, f: F) -> Scan<Self, S, Fut, F>
+    fn scan<'a, S: 'a, B, Fut, F>(self, initial_state: S, f: F) -> Scan<Self, S, Fut, F>
     where
-        F: FnMut(&mut S, Self::Item) -> Fut,
-        Fut: Future<Output = Option<B>>,
+        F: FnMut(&'a mut S, Self::Item) -> Fut,
+        Fut: Future<Output = Option<B>> + 'a,
         Self: Sized,
     {
         Scan::new(self, initial_state, f)

futures-util/src/stream/stream/scan.rs

@@ -6,6 +6,7 @@ use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use core::mem::transmute;
 
 struct StateFn<S, F> {
     state: S,
@@ -25,7 +26,6 @@ impl<St: Unpin + Stream, S, Fut: Unpin, F> Unpin for Scan<St, S, Fut, F> {}
 impl<St, S, Fut, F> fmt::Debug for Scan<St, S, Fut, F>
 where
     St: Stream + fmt::Debug,
-    St::Item: fmt::Debug,
     S: fmt::Debug,
     Fut: fmt::Debug,
 {
@@ -50,11 +50,11 @@ impl<St: Stream, S, Fut, F> Scan<St, S, Fut, F> {
     }
 }
 
-impl<B, St, S, Fut, F> Scan<St, S, Fut, F>
+impl<'a, B, St, S: 'a, Fut, F> Scan<St, S, Fut, F>
 where
     St: Stream,
-    F: FnMut(&mut S, St::Item) -> Fut,
-    Fut: Future<Output = Option<B>>,
+    F: FnMut(&'a mut S, St::Item) -> Fut,
+    Fut: Future<Output = Option<B>> + 'a,
 {
     pub(super) fn new(stream: St, initial_state: S, f: F) -> Scan<St, S, Fut, F> {
         Scan {
@@ -100,11 +100,11 @@ where
     }
 }
 
-impl<B, St, S, Fut, F> Stream for Scan<St, S, Fut, F>
+impl<'a, B, St, S: 'a, Fut, F> Stream for Scan<St, S, Fut, F>
 where
     St: Stream,
-    F: FnMut(&mut S, St::Item) -> Fut,
-    Fut: Future<Output = Option<B>>,
+    F: FnMut(&'a mut S, St::Item) -> Fut,
+    Fut: Future<Output = Option<B>> + 'a,
 {
     type Item = B;
 
@@ -119,7 +119,12 @@ where
                 None => return Poll::Ready(None),
             };
             let state_f = self.as_mut().state_f().as_mut().unwrap();
-            let fut = (state_f.f)(&mut state_f.state, item);
+            let fut = (state_f.f)(
+                // Safety: this's safe because state is internal and can only be accessed
+                // via provided function.
+                unsafe { transmute(&mut state_f.state) },
+                item,
+            );
             self.as_mut().future().set(Some(fut));
         }
 
@@ -142,11 +147,11 @@ where
     }
 }
 
-impl<B, St, S, Fut, F> FusedStream for Scan<St, S, Fut, F>
+impl<'a, B, St, S: 'a, Fut, F> FusedStream for Scan<St, S, Fut, F>
 where
     St: FusedStream,
-    F: FnMut(&mut S, St::Item) -> Fut,
-    Fut: Future<Output = Option<B>>,
+    F: FnMut(&'a mut S, St::Item) -> Fut,
+    Fut: Future<Output = Option<B>> + 'a,
 {
     fn is_terminated(&self) -> bool {
         self.is_done_taking() || self.future.is_none() && self.stream.is_terminated()

futures/tests/stream.rs

@@ -1,4 +1,5 @@
 use futures::executor::block_on;
+use futures::future::*;
 use futures::stream::{self, StreamExt};
 
 #[test]
@@ -15,18 +16,64 @@ fn select() {
     select_and_compare(vec![1, 2], vec![4, 5, 6], vec![1, 4, 2, 5, 6]);
 }
 
+async fn async_scan_fn(state: &mut u8, e: u8) -> Option<u8> {
+    *state += 1;
+    if e < *state {
+        Some(e)
+    } else {
+        None
+    }
+}
+
+fn impl_scan_fn<'a>(state: &'a mut u8, e: u8) -> impl Future<Output = Option<u8>> + 'a {
+    async move {
+        *state += 1;
+        if e < *state {
+            Some(e)
+        } else {
+            None
+        }
+    }
+}
+
 #[test]
 fn scan() {
     futures::executor::block_on(async {
         assert_eq!(
             stream::iter(vec![1u8, 2, 3, 4, 6, 8, 2])
-                .scan(1, |acc, e| {
-                    *acc += 1;
-                    futures::future::ready(if e < *acc { Some(e) } else { None })
+                .scan(1, |state, e| {
+                    *state += 1;
+                    futures::future::ready(if e < *state { Some(e) } else { None })
                 })
                 .collect::<Vec<_>>()
                 .await,
             vec![1u8, 2, 3, 4]
         );
     });
 }
+
+#[test]
+fn scan_with_async() {
+    futures::executor::block_on(async {
+        assert_eq!(
+            stream::iter(vec![1u8, 2, 3, 4, 6, 8, 2])
+                .scan(1u8, async_scan_fn)
+                .collect::<Vec<_>>()
+                .await,
+            vec![1u8, 2, 3, 4]
+        );
+    });
+}
+
+#[test]
+fn scan_with_impl() {
+    futures::executor::block_on(async {
+        assert_eq!(
+            stream::iter(vec![1u8, 2, 3, 4, 6, 8, 2])
+                .scan(1u8, impl_scan_fn)
+                .collect::<Vec<_>>()
+                .await,
+            vec![1u8, 2, 3, 4]
+        );
+    });
+}