Get server working again (mostly)

src/proto/connection.rs

@@ -1,4 +1,4 @@
-use {client, ConnectionError, Frame};
+use {client, server, ConnectionError, Frame};
 use HeaderMap;
 use frame::{self, StreamId};
 
@@ -122,19 +122,7 @@ impl<T, P, B> Connection<T, P, B>
             match frame {
                 Some(Headers(frame)) => {
                     trace!("recv HEADERS; frame={:?}", frame);
-
-                    if let Some(frame) = try!(self.streams.recv_headers::<P>(frame)) {
-                        unimplemented!();
-                    }
-
-                    /*
-                    // Update stream state while ensuring that the headers frame
-                    // can be received.
-                    if let Some(frame) = try!(self.streams.recv_headers(frame)) {
-                        let frame = Self::convert_poll_message(frame)?;
-                        return Ok(Some(frame).into());
-                    }
-                    */
+                    try!(self.streams.recv_headers::<P>(frame));
                 }
                 Some(Data(frame)) => {
                     trace!("recv DATA; frame={:?}", frame);
@@ -269,6 +257,15 @@ impl<T, B> Connection<T, client::Peer, B>
     }
 }
 
+impl<T, B> Connection<T, server::Peer, B>
+    where T: AsyncRead + AsyncWrite,
+          B: IntoBuf,
+{
+    pub fn next_incoming(&mut self) -> Option<StreamRef<B::Buf>> {
+        self.streams.next_incoming()
+    }
+}
+
 /*
 impl<T, B> Connection<T, Client, B>
     where T: AsyncRead + AsyncWrite,

src/proto/streams/recv.rs

@@ -1,4 +1,4 @@
-use {client, frame, ConnectionError};
+use {client, server, frame, ConnectionError};
 use proto::*;
 use super::*;
 
@@ -25,6 +25,9 @@ pub(super) struct Recv<B> {
     /// TODO: don't use a VecDeque
     pending_window_updates: VecDeque<StreamId>,
 
+    /// New streams to be accepted
+    pending_accept: store::List<B>,
+
     /// Holds frames that are waiting to be read
     buffer: Buffer<Bytes>,
 
@@ -54,6 +57,7 @@ impl<B> Recv<B> where B: Buf {
             init_window_sz: config.init_remote_window_sz,
             flow_control: FlowControl::new(config.init_remote_window_sz),
             pending_window_updates: VecDeque::new(),
+            pending_accept: store::List::new(),
             buffer: Buffer::new(),
             refused: None,
             _p: PhantomData,
@@ -78,6 +82,19 @@ impl<B> Recv<B> where B: Buf {
         Ok(Some(Stream::new(id)))
     }
 
+    pub fn take_request(&mut self, stream: &mut store::Ptr<B>)
+        -> Result<Request<()>, ConnectionError>
+    {
+        match stream.pending_recv.pop_front(&mut self.buffer) {
+            Some(Frame::Headers(frame)) => {
+                // TODO: This error should probably be caught on receipt of the
+                // frame vs. now.
+                Ok(server::Peer::convert_poll_message(frame)?)
+            }
+            _ => panic!(),
+        }
+    }
+
     pub fn poll_response(&mut self, stream: &mut store::Ptr<B>)
         -> Poll<Response<()>, ConnectionError> {
         // If the buffer is not empty, then the first frame must be a HEADERS
@@ -102,7 +119,7 @@ impl<B> Recv<B> where B: Buf {
     pub fn recv_headers<P: Peer>(&mut self,
                                  frame: frame::Headers,
                                  stream: &mut store::Ptr<B>)
-        -> Result<Option<frame::Headers>, ConnectionError>
+        -> Result<(), ConnectionError>
     {
         let is_initial = stream.state.recv_open(self.init_window_sz, frame.is_end_stream())?;
 
@@ -118,13 +135,14 @@ impl<B> Recv<B> where B: Buf {
         // Only servers can receive a headers frame that initiates the stream.
         // This is verified in `Streams` before calling this function.
         if P::is_server() {
-            Ok(Some(frame))
+            self.pending_accept.push(stream);
+            Ok(())
         } else {
             // Push the frame onto the recv buffer
             stream.pending_recv.push_back(&mut self.buffer, frame.into());
             stream.notify_recv();
 
-            Ok(None)
+            Ok(())
         }
     }
 
@@ -361,6 +379,10 @@ impl<B> Recv<B> where B: Buf {
         Ok(().into())
     }
 
+    pub fn next_incoming(&mut self, store: &mut Store<B>) -> Option<store::Key> {
+        self.pending_accept.pop(store)
+            .map(|ptr| ptr.key())
+    }
 
     pub fn poll_chunk(&mut self, stream: &mut Stream<B>)
         -> Poll<Option<Chunk>, ConnectionError>

src/proto/streams/streams.rs

@@ -1,4 +1,4 @@
-use client;
+use {client, server};
 use proto::*;
 use super::*;
 
@@ -63,7 +63,7 @@ impl<B> Streams<B>
 
     /// Process inbound headers
     pub fn recv_headers<P: Peer>(&mut self, frame: frame::Headers)
-        -> Result<Option<frame::Headers>, ConnectionError>
+        -> Result<(), ConnectionError>
     {
         let id = frame.stream_id();
         let mut me = self.inner.lock().unwrap();
@@ -82,7 +82,7 @@ impl<B> Streams<B>
 
                 match try!(me.actions.recv.open::<P>(id)) {
                     Some(stream) => e.insert(stream),
-                    None => return Ok(None),
+                    None => return Ok(()),
                 }
             }
         };
@@ -222,6 +222,22 @@ impl<B> Streams<B>
         */
     }
 
+    pub fn next_incoming(&mut self) -> Option<StreamRef<B>> {
+        let key = {
+            let mut me = self.inner.lock().unwrap();
+            let me = &mut *me;
+
+            me.actions.recv.next_incoming(&mut me.store)
+        };
+
+        key.map(|key| {
+            StreamRef {
+                inner: self.inner.clone(),
+                key,
+            }
+        })
+    }
+
     pub fn poll_window_update(&mut self)
         -> Poll<WindowUpdate, ConnectionError>
     {
@@ -335,6 +351,37 @@ impl<B> StreamRef<B>
         })
     }
 
+    /// Called by the server after the stream is accepted. Given that clients
+    /// initialize streams by sending HEADERS, the request will always be
+    /// available.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if the request isn't present.
+    pub fn take_request(&self) -> Result<Request<()>, ConnectionError> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+        me.actions.recv.take_request(&mut stream)
+    }
+
+    pub fn send_response(&mut self, response: Response<()>, end_of_stream: bool)
+        -> Result<(), ConnectionError>
+    {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.key);
+
+        let frame = server::Peer::convert_send_message(
+            stream.id, response, end_of_stream);
+
+        me.actions.transition::<server::Peer, _, _>(stream, |actions, stream| {
+            actions.send.send_headers(frame, stream)
+        })
+    }
+
     pub fn poll_response(&mut self) -> Poll<Response<()>, ConnectionError> {
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;

src/server.rs

@@ -4,7 +4,7 @@ use proto::{self, Connection};
 use error::Reason::*;
 
 use http::{self, Request, Response};
-use futures::{Future, Sink, Poll, Async, AsyncSink, IntoFuture};
+use futures::{self, Future, Sink, Poll, Async, AsyncSink, IntoFuture};
 use tokio_io::{AsyncRead, AsyncWrite};
 use bytes::{Bytes, IntoBuf};
 
@@ -86,6 +86,43 @@ impl<T, B> Server<T, B>
 
         Handshake { inner: Box::new(handshake) }
     }
+
+    pub fn poll_close(&mut self) -> Poll<(), ConnectionError> {
+        self.connection.poll()
+    }
+}
+
+impl<T, B> futures::Stream for Server<T, B>
+    where T: AsyncRead + AsyncWrite + 'static,
+          B: IntoBuf + 'static,
+{
+    type Item = (Request<Body<B>>, Stream<B>);
+    type Error = ConnectionError;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, ConnectionError> {
+        // Always try to advance the internal state. Getting NotReady also is
+        // needed to allow this function to return NotReady.
+        match self.poll_close()? {
+            Async::Ready(_) => {
+                // If the socket is closed, don't return anything
+                // TODO: drop any pending streams
+                return Ok(None.into())
+            }
+            _ => {}
+        }
+
+        if let Some(inner) = self.connection.next_incoming() {
+            let (head, _) = inner.take_request()?.into_parts();
+            let body = Body { inner: inner.clone() };
+
+            let request = Request::from_parts(head, body);
+            let incoming = Stream { inner };
+
+            return Ok(Some((request, incoming)).into());
+        }
+
+        Ok(Async::NotReady)
+    }
 }
 
 impl<T, B> fmt::Debug for Server<T, B>
@@ -100,6 +137,30 @@ impl<T, B> fmt::Debug for Server<T, B>
     }
 }
 
+// ===== impl Stream =====
+
+impl<B: IntoBuf> Stream<B> {
+    pub fn send_response(&mut self, response: Response<()>, end_of_stream: bool)
+        -> Result<(), ConnectionError>
+    {
+        self.inner.send_response(response, end_of_stream)
+    }
+
+    /// Send data
+    pub fn send_data(&mut self, data: B, end_of_stream: bool)
+        -> Result<(), ConnectionError>
+    {
+        self.inner.send_data::<Peer>(data.into_buf(), end_of_stream)
+    }
+
+    /// Send trailers
+    pub fn send_trailers(&mut self, trailers: ())
+        -> Result<(), ConnectionError>
+    {
+        unimplemented!();
+    }
+}
+
 // ===== impl Flush =====
 
 impl<T> Flush<T> {