Repository: actix/actix (language: rust)

Issue:
StreamThen combinator for ActorStream is broken
The `take` on the first line here:

https://github.com/actix/actix/blob/b7c7237223dae46d27426799a48b4354c4376f54/src/fut/stream_then.rs#L67-L73

Means that if the future doesn't complete immediately on first poll it will just be dropped. This is obviously incorrect behaviour: the future shouldn't be removed from the option until it return `Ready` from `poll`.


Discussion / hints:

Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 700f18ee8..949d2e6dc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -8,6 +8,12 @@
 
 [#365]: https://github.com/actix/actix/pull/365
 
+## Fixed
+
+* Fix `ActorFuture::poll_next` impl for `StreamThen` to not lose inner future when it's pending. [#376]
+
+[#376]: https://github.com/actix/actix/pull/376
+
 ## [0.10.0-alpha.2] 2020-03-05
 
 ## Added
diff --git a/src/fut/stream_then.rs b/src/fut/stream_then.rs
index 7ccda7c6d..d7052c36f 100644
--- a/src/fut/stream_then.rs
+++ b/src/fut/stream_then.rs
@@ -64,7 +64,7 @@ where
             this.future = Some((this.f)(item, act, ctx).into_future());
         }
         assert!(this.future.is_some());
-        match Pin::new(&mut this.future.take().unwrap()).poll(act, ctx, task) {
+        match Pin::new(this.future.as_mut().unwrap()).poll(act, ctx, task) {
             Poll::Ready(e) => {
                 this.future = None;
                 Poll::Ready(Some(e))


Test patch:
diff --git a/tests/test_fut.rs b/tests/test_fut.rs
index a093330fe..7732081a6 100644
--- a/tests/test_fut.rs
+++ b/tests/test_fut.rs
@@ -56,6 +56,16 @@ impl Actor for MyStreamActor {
 
         s.into_actor(self)
             .timeout(Duration::new(0, 1000))
+            .then(|res, act, _| {
+                // Additional waiting time to test `then` call as well
+                Box::pin(
+                    async move {
+                        delay_for(Duration::from_millis(500)).await;
+                        res
+                    }
+                    .into_actor(act),
+                )
+            })
             .map(|e, act, _| {
                 if let Err(()) = e {
                     act.timeout.store(true, Ordering::Relaxed);