From c23d11306ecc11f0bf583c626954660e4936ce59 Mon Sep 17 00:00:00 2001
From: Carl Lerche <me@carllerche.com>
Date: Sat, 21 Oct 2017 09:59:27 -0700
Subject: [PATCH] Add RecvStream::is_end_stream. (#165)

This function returns true if the `RecvStream` has reached an end of
stream state. This is intended to replace `is_empty` which has confusing
behavior.
---
 src/proto/streams/recv.rs    | 10 ++++++++++
 src/proto/streams/streams.rs |  9 +++++++++
 src/share.rs                 | 11 ++++++++++-
 tests/flow_control.rs        |  1 +
 tests/stream_states.rs       |  4 +++-
 5 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/src/proto/streams/recv.rs b/src/proto/streams/recv.rs
index 5d3e24b..b210ad6 100644
--- a/src/proto/streams/recv.rs
+++ b/src/proto/streams/recv.rs
@@ -330,6 +330,16 @@ impl Recv {
             .unwrap_or(true)
     }
 
+    pub fn is_end_stream(&self, stream: &store::Ptr) -> bool {
+        if !stream.state.is_recv_closed() {
+            return false;
+        }
+
+        stream
+            .pending_recv
+            .is_empty()
+    }
+
     pub fn recv_data(
         &mut self,
         frame: frame::Data,
diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs
index cc926fc..ad15047 100644
--- a/src/proto/streams/streams.rs
+++ b/src/proto/streams/streams.rs
@@ -758,6 +758,15 @@ impl OpaqueStreamRef {
         me.actions.recv.body_is_empty(&stream)
     }
 
+    pub fn is_end_stream(&self) -> bool {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.key);
+
+        me.actions.recv.is_end_stream(&stream)
+    }
+
     pub fn poll_data(&mut self) -> Poll<Option<Bytes>, proto::Error> {
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;
diff --git a/src/share.rs b/src/share.rs
index 98bf3cb..d60f4bc 100644
--- a/src/share.rs
+++ b/src/share.rs
@@ -70,12 +70,21 @@ impl RecvStream {
         RecvStream { inner }
     }
 
-    // TODO: Rename to "is_end_stream"
+    #[deprecated(since = "0.0.0")]
+    #[doc(hidden)]
     pub fn is_empty(&self) -> bool {
         // If the recv side is closed and the receive queue is empty, the body is empty.
         self.inner.inner.body_is_empty()
     }
 
+    /// Returns true if the receive half has reached the end of stream.
+    ///
+    /// A return value of `true` means that calls to `poll` and `poll_trailers`
+    /// will both return `None`.
+    pub fn is_end_stream(&self) -> bool {
+        self.inner.inner.is_end_stream()
+    }
+
     pub fn release_capacity(&mut self) -> &mut ReleaseCapacity {
         &mut self.inner
     }
diff --git a/tests/flow_control.rs b/tests/flow_control.rs
index d4464d0..44d7004 100644
--- a/tests/flow_control.rs
+++ b/tests/flow_control.rs
@@ -161,6 +161,7 @@ fn release_capacity_of_small_amount_does_not_send_window_update() {
                 .and_then(|resp| {
                     assert_eq!(resp.status(), StatusCode::OK);
                     let body = resp.into_parts().1;
+                    assert!(!body.is_end_stream());
                     body.into_future().unwrap()
                 })
                 // read the small body and then release it
diff --git a/tests/stream_states.rs b/tests/stream_states.rs
index 66397c3..f94185c 100644
--- a/tests/stream_states.rs
+++ b/tests/stream_states.rs
@@ -169,7 +169,9 @@ fn closed_streams_are_released() {
             // stream wired.
             assert_eq!(1, client.num_wired_streams());
 
-            drop(response);
+            let (_, body) = response.into_parts();
+            assert!(body.is_end_stream());
+            drop(body);
 
             // The stream state is now free
             assert_eq!(0, client.num_wired_streams());