wip: problems with Frame::len()

src/proto/connection.rs

@@ -1,23 +1,23 @@
 use {ConnectionError, Frame, FrameSize};
 use client::Client;
+use error;
 use frame::{self, SettingSet, StreamId};
 use proto::{self, ControlFlow, ControlPing, ControlSettings, Peer, PingPayload, ReadySink, WindowSize};
 use server::Server;
 
-use tokio_io::{AsyncRead, AsyncWrite};
-
-use http::{request, response};
 use bytes::{Bytes, IntoBuf};
-
+use http::{request, response};
 use futures::*;
-
+use tokio_io::{AsyncRead, AsyncWrite};
 use std::marker::PhantomData;
 
 /// An H2 connection
 #[derive(Debug)]
 pub struct Connection<T, P, B: IntoBuf = Bytes> {
     inner: proto::Transport<T, P, B::Buf>,
-    peer: PhantomData<P>,
+    // Set to `true` as long as the connection is in a valid state.
+    active: bool,
+    _phantom: PhantomData<(P, B)>,
 }
 
 pub fn new<T, P, B>(transport: proto::Transport<T, P, B::Buf>)
@@ -28,13 +28,14 @@ pub fn new<T, P, B>(transport: proto::Transport<T, P, B::Buf>)
 {
     Connection {
         inner: transport,
-        peer: PhantomData,
+        active: true,
+        _phantom: PhantomData,
     }
 }
 
 
 impl<T, P, B> ControlSettings for Connection<T, P, B>
-    where T: ControlSettings,
+    where T: AsyncRead + AsyncWrite,
           B: IntoBuf,
 {
     fn update_local_settings(&mut self, local: frame::SettingSet) -> Result<(), ConnectionError> {
@@ -51,7 +52,7 @@ impl<T, P, B> ControlSettings for Connection<T, P, B>
 }
 
 impl<T, P, B> ControlFlow for Connection<T, P, B>
-    where T: ControlFlow,
+    where T: AsyncRead + AsyncWrite,
           B: IntoBuf,
 {
     fn poll_remote_window_update(&mut self, id: StreamId) -> Poll<WindowSize, ConnectionError> {
@@ -65,7 +66,6 @@ impl<T, P, B> ControlFlow for Connection<T, P, B>
 
 impl<T, P, B> ControlPing for Connection<T, P, B>
     where T: AsyncRead + AsyncWrite,
-          T: ControlPing,
           P: Peer,
           B: IntoBuf,
 {
@@ -146,6 +146,10 @@ impl<T, P, B> Stream for Connection<T, P, B>
         use frame::Frame::*;
         trace!("poll");
 
+        if !self.active {
+            return Err(error::User::Corrupt.into());
+        }
+
         loop {
             let frame = match try!(self.inner.poll()) {
                 Async::Ready(f) => f,
@@ -153,7 +157,7 @@ impl<T, P, B> Stream for Connection<T, P, B>
                     // Receiving new frames may depend on ensuring that the write buffer
                     // is clear (e.g. if window updates need to be sent), so `poll_complete`
                     // is called here. 
-                    try_ready!(self.inner.poll_complete());
+                    try_ready!(self.poll_complete());
 
                     // If the write buffer is cleared, attempt to poll the underlying
                     // stream once more because it, may have been made ready.
@@ -172,7 +176,7 @@ impl<T, P, B> Stream for Connection<T, P, B>
                 Some(Data(v)) => Frame::Data {
                     id: v.stream_id(),
                     end_of_stream: v.is_end_stream(),
-                    data_len: v.len(),
+                    //data_len: v.len(),
                     data: v.into_payload(),
                 },
 
@@ -199,9 +203,13 @@ impl<T, P, B> Sink for Connection<T, P, B>
     {
         trace!("start_send");
 
+        if !self.active {
+            return Err(error::User::Corrupt.into());
+        }
+
         // Ensure the transport is ready to send a frame before we transform the external
-        // `Frame` into an internal `frame::Framme`.
-        if self.inner.poll_ready()? == Async::NotReady {
+        // `Frame` into an internal `frame::Frame`.
+        if !try!(self.poll_ready()).is_ready() {
             return Ok(AsyncSink::NotReady(item));
         }
 

src/proto/flow_control.rs

@@ -47,7 +47,7 @@ impl<T, U> FlowControl<T>
             initial_local_window_size,
             initial_remote_window_size,
             connection_local_flow_controller: FlowControlState::with_initial_size(initial_local_window_size),
-            connection_remote_flow_controller: FlowControlState::with_next_update(initial_remote_window_size),
+            connection_remote_flow_controller: FlowControlState::with_initial_size(initial_remote_window_size),
             blocked_remote_window_update: None,
             sending_local_window_update: None,
             pending_local_window_updates: VecDeque::new(),

src/proto/stream_tracker.rs

@@ -1,6 +1,6 @@
 use ConnectionError;
-use error::Reason::ProtocolError;
-use error::User::InvalidStreamId;
+use error::Reason;
+use error::User;
 use frame::{self, Frame};
 use proto::*;
 
@@ -125,7 +125,7 @@ impl<T, P> Stream for StreamTracker<T, P>
                     // connections should not be factored.
 
                     if !P::is_valid_remote_stream_id(id) {
-                        unimplemented!();
+                        return Err(Reason::ProtocolError.into());
                     }
                 }
 
@@ -134,7 +134,7 @@ impl<T, P> Stream for StreamTracker<T, P>
 
             Some(Data(v)) => {
                 match self.streams.get_mut(&v.stream_id()) {
-                    None => return Err(ProtocolError.into()),
+                    None => return Err(Reason::ProtocolError.into()),
                     Some(state) => state.recv_data(v.is_end_stream())?,
                 }
                 Ok(Async::Ready(Some(Data(v))))
@@ -179,14 +179,14 @@ impl<T, P, U> Sink for StreamTracker<T, P>
                     // connections should not be factored.
                     if !P::is_valid_local_stream_id(id) {
                         // TODO: clear state
-                        return Err(InvalidStreamId.into());
+                        return Err(User::InvalidStreamId.into());
                     }
                 }
             }
 
             &Data(ref v) => {
                 match self.streams.get_mut(&v.stream_id()) {
-                    None => return Err(ProtocolError.into()),
+                    None => return Err(User::InactiveStreamId.into()),
                     Some(state) => state.send_data(v.is_end_stream())?,
                 }
             }