Handle the remote returning a protocol error

src/proto/connection.rs

@@ -92,6 +92,16 @@ impl<T, P, B> Connection<T, P, B>
 
     /// Advances the internal state of the connection.
     pub fn poll(&mut self) -> Poll<(), ConnectionError> {
+        match self.poll2() {
+            Err(e) => {
+                self.streams.recv_err(&e);
+                Err(e)
+            }
+            ret => ret,
+        }
+    }
+
+    fn poll2(&mut self) -> Poll<(), ConnectionError> {
         use frame::Frame::*;
 
         loop {
@@ -178,9 +188,6 @@ impl<T, P, B> Connection<T, P, B>
                 }
             }
         }
-
-        // TODO: Flush the write buffer
-        unimplemented!();
     }
 
     /*

src/proto/streams/recv.rs

@@ -143,6 +143,14 @@ impl<P, B> Recv<P, B>
         Ok(())
     }
 
+    pub fn recv_err(&mut self, err: &ConnectionError, stream: &mut Stream<B>) {
+        // Receive an error
+        stream.state.recv_err(err);
+
+        // If a receiver is waiting, notify it
+        stream.notify_recv();
+    }
+
     pub fn dec_num_streams(&mut self) {
         self.num_streams -= 1;
     }
@@ -308,6 +316,8 @@ impl<B> Recv<client::Peer, B>
             }
             Some(frame) => unimplemented!(),
             None => {
+                stream.state.ensure_recv_open()?;
+
                 stream.recv_task = Some(task::current());
                 Ok(Async::NotReady)
             }

src/proto/streams/state.rs

@@ -66,7 +66,7 @@ enum Inner {
     HalfClosedLocal(Peer), // TODO: explicitly name this value
     HalfClosedRemote(Peer),
     // When reset, a reason is provided
-    Closed(Option<Reason>),
+    Closed(Option<Cause>),
 }
 
 #[derive(Debug, Copy, Clone)]
@@ -76,6 +76,12 @@ enum Peer {
     Streaming(FlowControl),
 }
 
+#[derive(Debug, Copy, Clone)]
+enum Cause {
+    Proto(Reason),
+    Io,
+}
+
 impl State {
     /// Opens the send-half of a stream if it is not already open.
     pub fn send_open(&mut self, sz: WindowSize, eos: bool) -> Result<(), ConnectionError> {
@@ -178,6 +184,19 @@ impl State {
         }
     }
 
+    pub fn recv_err(&mut self, err: &ConnectionError) {
+        match self.inner {
+            Closed(..) => {}
+            _ => {
+                self.inner = Closed(match *err {
+                    ConnectionError::Proto(reason) => Some(Cause::Proto(reason)),
+                    ConnectionError::Io(..) => Some(Cause::Io),
+                    _ => panic!("cannot terminate stream with user error"),
+                });
+            }
+        }
+    }
+
     /// Indicates that the local side will not send more data to the local.
     pub fn send_close(&mut self) -> Result<(), ConnectionError> {
         match self.inner {
@@ -225,6 +244,21 @@ impl State {
             _ => None,
         }
     }
+
+    pub fn ensure_recv_open(&self) -> Result<(), ConnectionError> {
+        use std::io;
+
+        // TODO: Is this correct?
+        match self.inner {
+            Closed(Some(Cause::Proto(reason))) => {
+                Err(ConnectionError::Proto(reason))
+            }
+            Closed(Some(Cause::Io)) => {
+                Err(ConnectionError::Io(io::ErrorKind::BrokenPipe.into()))
+            }
+            _ => Ok(()),
+        }
+    }
 }
 
 impl Default for State {

src/proto/streams/store.rs

@@ -90,6 +90,14 @@ impl<B> Store<B> {
             }
         }
     }
+
+    pub fn for_each<F>(&mut self, mut f: F)
+        where F: FnMut(&mut Stream<B>)
+    {
+        for &id in self.ids.values() {
+            f(&mut self.slab[id])
+        }
+    }
 }
 
 // ===== impl Ptr =====
@@ -142,7 +150,7 @@ impl<'a, B> OccupiedEntry<'a, B> {
 }
 
 // ===== impl VacantEntry =====
-//
+
 impl<'a, B> VacantEntry<'a, B> {
     pub fn insert(self, value: Stream<B>) -> Key {
         // Insert the value in the slab

src/proto/streams/streams.rs

@@ -141,6 +141,14 @@ impl<P, B> Streams<P, B>
         unimplemented!();
     }
 
+    pub fn recv_err(&mut self, err: &ConnectionError) {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let actions = &mut me.actions;
+        me.store.for_each(|stream| actions.recv.recv_err(err, stream));
+    }
+
     pub fn recv_window_update(&mut self, frame: frame::WindowUpdate)
         -> Result<(), ConnectionError> {
         let id = frame.stream_id();