implement h2::server::Stream::send_reset(Reason) and Body::is_empty()

src/lib.rs

@@ -60,6 +60,13 @@ pub struct Chunk<B: IntoBuf> {
 
 // ===== impl Body =====
 
+impl<B: IntoBuf> Body<B> {
+    pub fn is_empty(&self) -> bool {
+        // If the recv side is closed and the receive queue is empty, the body is empty.
+        self.inner.is_recv_eos()
+    }
+}
+
 impl<B: IntoBuf> futures::Stream for Body<B> {
     type Item = Chunk<B>;
     type Error = ConnectionError;

src/proto/streams/send.rs

@@ -1,4 +1,5 @@
 use {frame, ConnectionError};
+use error::User::InactiveStreamId;
 use proto::*;
 use super::*;
 
@@ -107,6 +108,22 @@ impl<B> Send<B> where B: Buf {
         stream.state.send_close()
     }
 
+    pub fn send_reset(&mut self, reason: Reason,
+                      stream: &mut store::Ptr<B>)
+        -> Result<(), ConnectionError>
+    {
+        if stream.state.is_closed() {
+            return Err(InactiveStreamId.into())
+        }
+
+        stream.state.send_reset(reason)?;
+
+        let frame = frame::Reset::new(stream.id, reason);
+        self.prioritize.queue_frame(frame.into(), stream);
+
+        Ok(())
+    }
+
     pub fn send_data(&mut self,
                      frame: frame::Data<B>,
                      stream: &mut store::Ptr<B>)

src/proto/streams/state.rs

@@ -242,6 +242,19 @@ impl State {
         }
     }
 
+    /// Indicates that the local side will not send more data to the local.
+    pub fn send_reset(&mut self, reason: Reason) -> Result<(), ConnectionError> {
+        match self.inner {
+            Idle => Err(ProtocolError.into()),
+            Closed(..) => Ok(()),
+            _ => {
+                trace!("send_reset: => Closed");
+                self.inner = Closed(Some(Cause::Proto(reason)));
+                Ok(())
+            }
+        }
+    }
+
     /// Returns true if a stream with the current state counts against the
     /// concurrency limit.
     pub fn is_counted(&self) -> bool {

src/proto/streams/streams.rs

@@ -336,6 +336,16 @@ impl<B> StreamRef<B>
         me.actions.recv.take_request(&mut stream)
     }
 
+    pub fn send_reset<P: Peer>(&mut self, reason: Reason) -> Result<(), ConnectionError> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.key);
+        me.actions.transition::<P, _, _>(stream, move |actions, stream| {
+            actions.send.send_reset(reason, stream)
+        })
+    }
+
     pub fn send_response(&mut self, response: Response<()>, end_of_stream: bool)
         -> Result<(), ConnectionError>
     {
@@ -352,6 +362,15 @@ impl<B> StreamRef<B>
         })
     }
 
+    pub fn is_recv_eos(&self) -> bool {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.key);
+
+        stream.state.is_recv_closed() && stream.pending_recv.is_empty()
+    }
+
     pub fn poll_response(&mut self) -> Poll<Response<()>, ConnectionError> {
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;

src/server.rs

@@ -1,12 +1,12 @@
-use {frame, ConnectionError, StreamId};
+use {frame, ConnectionError, Reason, StreamId};
 use {Body, Chunk};
 use proto::{self, Connection, WindowSize};
 use error::Reason::*;
 
 use http::{self, Request, Response};
 use futures::{self, Future, Sink, Poll, Async, AsyncSink, IntoFuture};
 use tokio_io::{AsyncRead, AsyncWrite};
-use bytes::{Bytes, IntoBuf};
+use bytes::{Bytes, IntoBuf, Buf};
 
 use std::fmt;
 
@@ -191,6 +191,10 @@ impl<B: IntoBuf> Stream<B> {
     {
         unimplemented!();
     }
+
+    pub fn send_reset(mut self, reason: Reason) -> Result<(), ConnectionError> {
+        self.inner.send_reset::<Peer>(reason)
+    }
 }
 
 impl Stream<Bytes> {