Repository: actix/actix (language: rust)

Issue:
Future returned from Message Handler is not executed if Message is sent via Context::add_message_stream
## MRE

deps:
```
actix = "0.9"
futures = "0.3"
```

```rust
use actix::{Actor, Context, StreamHandler, Message, AsyncContext, Handler, ResponseFuture, System};
use futures::future::FutureExt;

struct Foo{}

impl Actor for Foo {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.add_message_stream(futures::stream::once(async {
            Msg {}
        }))
    }
}

struct Msg {}

impl Message for Msg {
    type Result = Result<(),()>;
}

impl Handler<Msg> for Foo {
    type Result = ResponseFuture<Result<(),()>>;

    fn handle(&mut self, _: Msg, _: &mut Self::Context) -> Self::Result {
        async {
            unreachable!();
            Ok(())
        }.boxed_local()
    }
}

fn main() {
    let sys = System::new("system");
    let actor = Foo {}.start();
    sys.run().unwrap()
}
```

## Expected behaviour

Arbiter thread panics at `unreachable!()` statement.

## Actual behaviour

`unreachable!()` statement is not being executed.

## Additional info

actix v0.8 works as expected executing returned future and triggering panic.

Sending message via Address::send:
```
    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.address().send(Msg {}).map(|_|()).into_actor(self).spawn(ctx);
    }
```
triggers panic.

Actually, im not sure that this is a bug, new behaviour kinda makes sense :)


Discussion / hints:
> new behaviour kinda makes sense :)

I'm unsure about that. It's quite subtle.

The regression actually appeared here:  
https://github.com/actix/actix/commit/aeb7bd15ea8793e5f00f17caefb01eb678136c2e#diff-057239071828d04f3e4c9ce9b814cdb0L179-L185  
where
```rust
actix_rt::spawn(self.then(move |res| {
    if let Some(tx) = tx {
        tx.send(res)
    }
```
was replaced with
```rust
actix_rt::spawn(async move {
    if let Some(tx) = tx {
        tx.send(self.await)
    }
```
so the `Stream` is polled only if `tx` is provided.


Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 14a78b23c..edc9898bc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,11 @@
 # CHANGES
 
+## [0.9.1] 2020-??-??
+
+### Fixed
+
+* Fix `MessageResponse` implementation  for `ResponseFuture` to always poll the spawned `Future`. (#317)
+
 ## [0.9.0] 2019-12-20
 
 ### Fixed
diff --git a/src/handler.rs b/src/handler.rs
index 7a671d3a0..03ce8fdf6 100644
--- a/src/handler.rs
+++ b/src/handler.rs
@@ -180,8 +180,9 @@ where
 {
     fn handle<R: ResponseChannel<M>>(self, _: &mut A::Context, tx: Option<R>) {
         actix_rt::spawn(async move {
+            let res = self.await;
             if let Some(tx) = tx {
-                tx.send(self.await)
+                tx.send(res)
             }
         });
     }


Test patch: