fix Body to return errors when there is recv error

src/proto/streams/recv.rs

@@ -565,16 +565,7 @@ where
                 // No more data frames
                 Ok(None.into())
             },
-            None => {
-                if stream.state.is_recv_closed() {
-                    // No more data frames will be received
-                    Ok(None.into())
-                } else {
-                    // Request to get notified once more data frames arrive
-                    stream.recv_task = Some(task::current());
-                    Ok(Async::NotReady)
-                }
-            },
+            None => self.schedule_recv(stream),
         }
     }
 
@@ -590,16 +581,18 @@ where
                 // we do?
                 unimplemented!();
             },
-            None => {
-                if stream.state.is_recv_closed() {
-                    // There will be no trailer frame
-                    Ok(None.into())
-                } else {
-                    // Request to get notified once another frame arrives
-                    stream.recv_task = Some(task::current());
-                    Ok(Async::NotReady)
-                }
-            },
+            None => self.schedule_recv(stream),
+        }
+    }
+
+    fn schedule_recv<T>(&mut self, stream: &mut Stream<B, P>) -> Poll<Option<T>, proto::Error> {
+        if stream.state.ensure_recv_open()? {
+            // Request to get notified once more frames arrive
+            stream.recv_task = Some(task::current());
+            Ok(Async::NotReady)
+        } else {
+            // No more frames will be received
+            Ok(None.into())
         }
     }
 }

src/proto/streams/state.rs

@@ -314,14 +314,15 @@ impl State {
         }
     }
 
-    pub fn ensure_recv_open(&self) -> Result<(), proto::Error> {
+    pub fn ensure_recv_open(&self) -> Result<bool, proto::Error> {
         use std::io;
 
         // TODO: Is this correct?
         match self.inner {
             Closed(Some(Cause::Proto(reason))) => Err(proto::Error::Proto(reason)),
             Closed(Some(Cause::Io)) => Err(proto::Error::Io(io::ErrorKind::BrokenPipe.into())),
-            _ => Ok(()),
+            Closed(None) | HalfClosedRemote(..) => Ok(false),
+            _ => Ok(true),
         }
     }
 }

tests/flow_control.rs

@@ -225,17 +225,8 @@ fn recv_data_overflows_connection_window() {
             .and_then(|resp| {
                 assert_eq!(resp.status(), StatusCode::OK);
                 let body = resp.into_parts().1;
+                // FIXME: body stream should error also
                 body.concat2().unwrap()
-                /* FIXME: body stream should error also
-                        .then(|res| {
-                            let err = res.unwrap_err();
-                            assert_eq!(
-                                err.to_string(),
-                                "protocol error: flow-control protocol violated"
-                            );
-                            Ok::<(), ()>(())
-                        })
-                        */
             });
 
         // client should see a flow control error