fixup! Properly propagate substream state to caller

muxers/mplex/src/io.rs

@@ -798,14 +798,6 @@ where
         id
     }
 
-    /// Checks whether a substream is open for reading.
-    fn can_read(&self, id: &LocalStreamId) -> bool {
-        matches!(
-            self.substreams.get(id),
-            Some(SubstreamState::Open { .. }) | Some(SubstreamState::SendClosed { .. })
-        )
-    }
-
     /// Sends pending frames, without flushing.
     fn send_pending_frames(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         while let Some(frame) = self.pending_frames.pop_back() {
@@ -1266,33 +1258,26 @@ mod tests {
                     }
                 }
 
-                // Drain the buffer by reading from the stream.
-                for _ in 0..cfg.max_buffer_len + 1 {
-                    match m.poll_read_stream(cx, id) {
-                        Poll::Ready(Ok(Some(bytes))) => {
-                            assert_eq!(bytes, data);
-                        }
-                        x => panic!("Unexpected: {:?}", x),
-                    }
-                }
-
-                // Read from the stream after the buffer has been drained,
-                // expecting either EOF or further data, depending on
-                // the `MaxBufferBehaviour`.
+                // Try to read from the stream:
+                // - Resets are prioritized over data frames so we receive a reset right away.
+                // - If our behaviour is "block", we should be able to read all the data.
                 match cfg.max_buffer_behaviour {
                     MaxBufferBehaviour::ResetStream => {
                         // Expect to read EOF
                         match m.poll_read_stream(cx, id) {
-                            Poll::Ready(Ok(None)) => {}
+                            Poll::Ready(Err(e)) if e.kind() == ErrorKind::ConnectionReset => {}
                             poll => panic!("Unexpected: {:?}", poll),
                         }
                     }
                     MaxBufferBehaviour::Block => {
-                        // Expect to be able to continue reading.
-                        match m.poll_read_stream(cx, id) {
-                            Poll::Ready(Ok(Some(bytes))) => assert_eq!(bytes, data),
-                            Poll::Pending => assert_eq!(overflow.get(), 1),
-                            poll => panic!("Unexpected: {:?}", poll),
+                        for _ in 0..cfg.max_buffer_len + 2 {
+                            match m.poll_read_stream(cx, id) {
+                                Poll::Ready(Ok(Some(bytes))) => {
+                                    assert_eq!(bytes, data);
+                                }
+                                Poll::Pending => assert_eq!(overflow.get(), 1),
+                                poll => panic!("Unexpected: {:?}", poll),
+                            }
                         }
                     }
                 }