forked/h2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update comments

Browse Source
This commit is contained in:
Oliver Gould
2017-07-23 17:58:37 +00:00
parent 4f723fffce
commit 82ba0dde71
9 changed files with 228 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/proto/flow_control.rs
View File

@@ -173,6 +173,10 @@ impl<T, U> FlowControl<T>
}
}
/// Tracks window updates received from the remote and ensures that the remote does not
/// violate the local peer's flow controller.
///
/// TODO send flow control reset when the peer violates the flow control window.
impl<T> Stream for FlowControl<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: ControlStreams,
@@ -201,6 +205,7 @@ impl<T> Stream for FlowControl<T>
// controller. That's fine.
if let Some(fc) = self.recv_flow_controller(v.stream_id()) {
if fc.claim_window(sz).is_err() {
// TODO send flow control reset.
return Err(error::Reason::FlowControlError.into())
}
}
@@ -213,6 +218,12 @@ impl<T> Stream for FlowControl<T>
}
}
/// Tracks the send flow control windows for sent frames.
///
/// If sending a frame would violate the remote's window, start_send fails with
/// `FlowControlViolation`.
///
/// Sends pending window updates before operating on the underlying transport.
impl<T, U> Sink for FlowControl<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
@@ -270,6 +281,7 @@ impl<T, U> Sink for FlowControl<T>
}
}
/// Sends pending window updates before checking the underyling transport's readiness.
impl<T, U> ReadySink for FlowControl<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
@@ -331,6 +343,7 @@ impl<T> ApplySettings for FlowControl<T>
}
}
/// Proxy.
impl<T: ControlStreams> ControlStreams for FlowControl<T> {
fn local_valid_id(id: StreamId) -> bool {
T::local_valid_id(id)
@@ -375,6 +388,7 @@ impl<T: ControlStreams> ControlStreams for FlowControl<T> {
fn get_reset(&self, id: StreamId) -> Option<Reason> {
self.inner.get_reset(id)
}
fn is_local_active(&self, id: StreamId) -> bool {
self.inner.is_local_active(id)
}
@@ -391,11 +405,11 @@ impl<T: ControlStreams> ControlStreams for FlowControl<T> {
self.inner.remote_active_len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_recv_window_size(old_sz, new_sz)
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_send_window_size(old_sz, new_sz)
}
@@ -407,15 +421,16 @@ impl<T: ControlStreams> ControlStreams for FlowControl<T> {
self.inner.send_flow_controller(id)
}
fn can_send_data(&mut self, id: StreamId) -> bool {
self.inner.can_send_data(id)
fn is_send_open(&mut self, id: StreamId) -> bool {
self.inner.is_send_open(id)
}
fn can_recv_data(&mut self, id: StreamId) -> bool {
self.inner.can_recv_data(id)
fn is_recv_open(&mut self, id: StreamId) -> bool {
self.inner.is_recv_open(id)
}
}
/// Proxy.
impl<T: ControlPing> ControlPing for FlowControl<T> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)

41
src/proto/mod.rs
View File

@@ -53,7 +53,28 @@ use self::stream_store::{ControlStreams, StreamStore};
/// - Exposes ControlSettings up towards the application and transmits local settings to
/// the remote.
///
/// ### `FlowControl`
/// ### The stream transport
///
/// The states of all HTTP/2 connections are stored centrally in the `StreamStore` at the
/// bottom of the stream transport. Several modules above this access this state via the
/// `ControlStreams` API to drive changes to the stream state. In each direction (send
/// from local to remote, and recv from remote to local), there is an Stream\*Open module
/// responsible for initializing new streams and ensuring that frames do not violate
/// stream state. Then, there are modules that operate on streams (for instance,
/// FlowControl). Finally, a Stream\*Close module is responsible for acting on END_STREAM
/// frames to ensure that stream states are not closed before work is complete.
///
/// #### `StreamSendOpen`
///
/// - Initializes streams initiated by the local peer.
/// - Ensures that frames sent from the local peer are appropriate for the stream's state.
/// - Ensures that the remote's max stream concurrency is not violated.
///
/// #### `StreamRecvClose`
///
/// - Updates the stream state for frames sent with END_STREAM.
///
/// #### `FlowControl`
///
/// - Tracks received data frames against the local stream and connection flow control
/// windows.
@@ -66,10 +87,22 @@ use self::stream_store::{ControlStreams, StreamStore};
/// - Sends window updates for the local stream and connection flow control windows as
/// instructed by upper layers.
///
/// ### `StreamTracker`
/// #### `StreamSendClose`
///
/// - Tracks all active streams.
/// - Tracks all reset streams.
/// - Updates the stream state for frames receive` with END_STREAM.
///
/// #### `StreamRecvOpen`
///
/// - Initializes streams initiated by the remote peer.
/// - Ensures that frames received from the remote peer are appropriate for the stream's
/// state.
/// - Ensures that the local peer's max stream concurrency is not violated.
/// - Emits StreamRefused resets to the remote.
///
/// #### `StreamStore`
///
/// - Holds the state of all local & remote active streams.
/// - Holds the cause of all reset/closed streams.
/// - Exposes `ControlStreams` so that upper layers may share stream state.
///
/// ### `PingPong`

4
src/proto/state.rs
View File

@@ -155,7 +155,7 @@ impl StreamState {
Ok(true)
}
pub fn can_send_data(&self) -> bool {
pub fn is_send_open(&self) -> bool {
use self::StreamState::*;
match self {
&Idle | &Closed | &HalfClosedRemote(..) => false,
@@ -165,7 +165,7 @@ impl StreamState {
}
}
pub fn can_recv_data(&self) -> bool {
pub fn is_recv_open(&self) -> bool {
use self::StreamState::*;
match self {
&Idle | &Closed | &HalfClosedLocal(..) => false,

26
src/proto/stream_recv_close.rs
View File

@@ -4,12 +4,7 @@ use frame::{self, Frame};
use proto::*;
use proto::ready::ReadySink;
// TODO track "last stream id" for GOAWAY.
// TODO track/provide "next" stream id.
// TODO reset_streams needs to be bounded.
// TODO track reserved streams (PUSH_PROMISE).
/// Handles end-of-stream frames sent from the remote.
/// Tracks END_STREAM frames received from the remote peer.
#[derive(Debug)]
pub struct StreamRecvClose<T> {
inner: T,
@@ -25,6 +20,7 @@ impl<T, U> StreamRecvClose<T>
}
}
/// Tracks END_STREAM frames received from the remote peer.
impl<T> Stream for StreamRecvClose<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: ControlStreams,
@@ -55,6 +51,7 @@ impl<T> Stream for StreamRecvClose<T>
}
}
// Proxy.
impl<T, U> Sink for StreamRecvClose<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ControlStreams,
@@ -71,6 +68,7 @@ impl<T, U> Sink for StreamRecvClose<T>
}
}
// Proxy.
impl<T, U> ReadySink for StreamRecvClose<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
@@ -81,6 +79,7 @@ impl<T, U> ReadySink for StreamRecvClose<T>
}
}
// Proxy.
impl<T: ControlStreams> ControlStreams for StreamRecvClose<T> {
fn local_valid_id(id: StreamId) -> bool {
T::local_valid_id(id)
@@ -142,11 +141,11 @@ impl<T: ControlStreams> ControlStreams for StreamRecvClose<T> {
self.inner.remote_active_len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_recv_window_size(old_sz, new_sz)
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_send_window_size(old_sz, new_sz)
}
@@ -158,15 +157,16 @@ impl<T: ControlStreams> ControlStreams for StreamRecvClose<T> {
self.inner.send_flow_controller(id)
}
fn can_send_data(&mut self, id: StreamId) -> bool {
self.inner.can_send_data(id)
fn is_send_open(&mut self, id: StreamId) -> bool {
self.inner.is_send_open(id)
}
fn can_recv_data(&mut self, id: StreamId) -> bool {
self.inner.can_recv_data(id)
fn is_recv_open(&mut self, id: StreamId) -> bool {
self.inner.is_recv_open(id)
}
}
// Proxy.
impl<T: ApplySettings> ApplySettings for StreamRecvClose<T> {
fn apply_local_settings(&mut self, set: &frame::SettingSet) -> Result<(), ConnectionError> {
self.inner.apply_local_settings(set)
@@ -177,6 +177,7 @@ impl<T: ApplySettings> ApplySettings for StreamRecvClose<T> {
}
}
// Proxy.
impl<T: ControlFlow> ControlFlow for StreamRecvClose<T> {
fn poll_window_update(&mut self) -> Poll<WindowUpdate, ConnectionError> {
self.inner.poll_window_update()
@@ -187,6 +188,7 @@ impl<T: ControlFlow> ControlFlow for StreamRecvClose<T> {
}
}
// Proxy.
impl<T: ControlPing> ControlPing for StreamRecvClose<T> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)

151
src/proto/stream_recv_open.rs
View File

@@ -3,7 +3,7 @@ use error::Reason::{ProtocolError, RefusedStream};
use frame::{Frame, StreamId};
use proto::*;
/// Tracks a connection's streams.
/// Ensures that frames are received on open streams in the appropriate state.
#[derive(Debug)]
pub struct StreamRecvOpen<T> {
inner: T,
@@ -60,25 +60,7 @@ impl<T, U> StreamRecvOpen<T>
}
}
/// Handles updates to `SETTINGS_MAX_CONCURRENT_STREAMS`.
///
/// > Indicates the maximum number of concurrent streams that the senderg will allow. This
/// > limit is directional: it applies to the number of streams that the sender permits
/// > the receiver to create. Initially, there is no limit to this value. It is
/// > recommended that this value be no smaller than 100, so as to not unnecessarily limit
/// > parallelism.
/// >
/// > A value of 0 for SETTINGS_MAX_CONCURRENT_STREAMS SHOULD NOT be treated as special by
/// > endpoints. A zero value does prevent the creation of new streams; however, this can
/// > also happen for any limit that is exhausted with active streams. Servers SHOULD only
/// > set a zero value for short durations; if a server does not wish to accept requests,
/// > closing the connection is more appropriate.
///
/// > An endpoint that wishes to reduce the value of SETTINGS_MAX_CONCURRENT_STREAMS to a
/// > value that is below the current number of open streams can either close streams that
/// > exceed the new value or allow streams to complete.
///
/// This module does NOT close streams when the setting changes.
/// Handles updates to `SETTINGS_MAX_CONCURRENT_STREAMS` from the local peer.
impl<T> ApplySettings for StreamRecvOpen<T>
where T: ApplySettings
{
@@ -93,6 +75,7 @@ impl<T> ApplySettings for StreamRecvOpen<T>
}
}
/// Helper.
impl<T: ControlStreams> StreamRecvOpen<T> {
fn check_not_reset(&self, id: StreamId) -> Result<(), ConnectionError> {
// Ensure that the stream hasn't been closed otherwise.
@@ -103,6 +86,7 @@ impl<T: ControlStreams> StreamRecvOpen<T> {
}
}
/// Ensures that frames are received on open streams in the appropriate state.
impl<T, U> Stream for StreamRecvOpen<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
@@ -112,8 +96,6 @@ impl<T, U> Stream for StreamRecvOpen<T>
type Error = T::Error;
fn poll(&mut self) -> Poll<Option<T::Item>, T::Error> {
use frame::Frame::*;
// Since there's only one slot for pending refused streams, it must be cleared
// before polling a frame from the transport.
try_ready!(self.send_pending_refuse());
@@ -142,6 +124,7 @@ impl<T, U> Stream for StreamRecvOpen<T>
&Frame::Headers(..) => {
self.check_not_reset(id)?;
if T::remote_valid_id(id) {
if self.inner.is_remote_active(id) {
// Can't send a a HEADERS frame on a remote stream that's
@@ -174,7 +157,7 @@ impl<T, U> Stream for StreamRecvOpen<T>
// All other stream frames are sent only when
_ => {
self.check_not_reset(id)?;
if !self.inner.can_recv_data(id) {
if !self.inner.is_recv_open(id) {
return Err(ProtocolError.into());
}
}
@@ -187,7 +170,7 @@ impl<T, U> Stream for StreamRecvOpen<T>
}
}
/// Ensures that a pending reset is
/// Sends pending resets before operating on the underlying transport.
impl<T, U> Sink for StreamRecvOpen<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ControlStreams,
@@ -211,7 +194,7 @@ impl<T, U> Sink for StreamRecvOpen<T>
}
}
/// Sends pending resets before checking the underlying transport's readiness.
impl<T, U> ReadySink for StreamRecvOpen<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
@@ -227,110 +210,7 @@ impl<T, U> ReadySink for StreamRecvOpen<T>
}
}
// Some(Headers(v)) => {
// let id = v.stream_id();
// let eos = v.is_end_stream();
// if self.get_reset(id).is_some() {
// // TODO send the remote errors when it sends us frames on reset
// // streams.
// continue;
// }
// if let Some(mut s) = self.get_active_mut(id) {
// let created = s.recv_headers(eos, self.initial_window_size)?;
// assert!(!created);
// return Ok(Async::Ready(Some(Headers(v))));
// }
// // Ensure that receiving this frame will not violate the local max
// // stream concurrency setting. Ensure that the stream is refused
// // before processing additional frames.
// if let Some(max) = self.max_concurrency {
// let max = max as usize;
// if !self.local.is_active(id) && self.local.active_count() >= max - 1 {
// // This frame would violate our local max concurrency, so reject
// // the stream.
// try_ready!(self.send_refusal(id));
// // Try to process another frame (hopefully for an active
// // stream).
// continue;
// }
// }
// let is_closed = {
// let stream = self.active_streams.entry(id)
// .or_insert_with(|| StreamState::default());
// let initialized =
// stream.recv_headers(eos, self.initial_window_size)?;
// if initialized {
// if !P::is_valid_remote_stream_id(id) {
// return Err(Reason::ProtocolError.into());
// }
// }
// stream.is_closed()
// };
// if is_closed {
// self.active_streams.remove(id);
// self.reset_streams.insert(id, Reason::NoError);
// }
// return Ok(Async::Ready(Some(Headers(v))));
// }
// Some(Data(v)) => {
// let id = v.stream_id();
// if self.get_reset(id).is_some() {
// // TODO send the remote errors when it sends us frames on reset
// // streams.
// continue;
// }
// let is_closed = {
// let stream = match self.active_streams.get_mut(id) {
// None => return Err(Reason::ProtocolError.into()),
// Some(s) => s,
// };
// stream.recv_data(v.is_end_stream())?;
// stream.is_closed()
// };
// if is_closed {
// self.reset(id, Reason::NoError);
// }
// return Ok(Async::Ready(Some(Data(v))));
// }
// Some(Reset(v)) => {
// // Set or update the reset reason.
// self.reset(v.stream_id(), v.reason());
// return Ok(Async::Ready(Some(Reset(v))));
// }
// Some(f) => {
// let id = f.stream_id();
// if self.get_reset(id).is_some() {
// // TODO send the remote errors when it sends us frames on reset
// // streams.
// continue;
// }
// return Ok(Async::Ready(Some(f)));
// }
// None => {
// return Ok(Async::Ready(None));
// }
/// Proxy.
impl<T: ControlStreams> ControlStreams for StreamRecvOpen<T> {
fn local_valid_id(id: StreamId) -> bool {
T::local_valid_id(id)
@@ -392,11 +272,11 @@ impl<T: ControlStreams> ControlStreams for StreamRecvOpen<T> {
self.inner.remote_active_len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_recv_window_size(old_sz, new_sz)
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_send_window_size(old_sz, new_sz)
}
@@ -408,15 +288,16 @@ impl<T: ControlStreams> ControlStreams for StreamRecvOpen<T> {
self.inner.send_flow_controller(id)
}
fn can_send_data(&mut self, id: StreamId) -> bool {
self.inner.can_send_data(id)
fn is_send_open(&mut self, id: StreamId) -> bool {
self.inner.is_send_open(id)
}
fn can_recv_data(&mut self, id: StreamId) -> bool {
self.inner.can_recv_data(id)
fn is_recv_open(&mut self, id: StreamId) -> bool {
self.inner.is_recv_open(id)
}
}
/// Proxy.
impl<T: ControlPing> ControlPing for StreamRecvOpen<T> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)

24
src/proto/stream_send_close.rs
View File

@@ -3,11 +3,7 @@ use error::Reason;
use frame::{self, Frame};
use proto::*;
// TODO track "last stream id" for GOAWAY.
// TODO track/provide "next" stream id.
// TODO reset_streams needs to be bounded.
// TODO track reserved streams (PUSH_PROMISE).
/// Tracks END_STREAM frames sent from the local peer.
#[derive(Debug)]
pub struct StreamSendClose<T> {
inner: T,
@@ -23,6 +19,7 @@ impl<T, U> StreamSendClose<T>
}
}
/// Proxy.
impl<T> Stream for StreamSendClose<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: ControlStreams,
@@ -35,6 +32,7 @@ impl<T> Stream for StreamSendClose<T>
}
}
/// Tracks END_STREAM frames sent from the local peer.
impl<T, U> Sink for StreamSendClose<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ControlStreams,
@@ -65,6 +63,7 @@ impl<T, U> Sink for StreamSendClose<T>
}
}
/// Proxy.
impl<T, U> ReadySink for StreamSendClose<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
@@ -75,6 +74,7 @@ impl<T, U> ReadySink for StreamSendClose<T>
}
}
/// Proxy.
impl<T: ApplySettings> ApplySettings for StreamSendClose<T> {
fn apply_local_settings(&mut self, set: &frame::SettingSet) -> Result<(), ConnectionError> {
self.inner.apply_local_settings(set)
@@ -85,6 +85,7 @@ impl<T: ApplySettings> ApplySettings for StreamSendClose<T> {
}
}
/// Proxy.
impl<T: ControlStreams> ControlStreams for StreamSendClose<T> {
fn local_valid_id(id: StreamId) -> bool {
T::local_valid_id(id)
@@ -146,11 +147,11 @@ impl<T: ControlStreams> ControlStreams for StreamSendClose<T> {
self.inner.remote_active_len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_recv_window_size(old_sz, new_sz)
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_send_window_size(old_sz, new_sz)
}
@@ -162,15 +163,16 @@ impl<T: ControlStreams> ControlStreams for StreamSendClose<T> {
self.inner.send_flow_controller(id)
}
fn can_send_data(&mut self, id: StreamId) -> bool {
self.inner.can_send_data(id)
fn is_send_open(&mut self, id: StreamId) -> bool {
self.inner.is_send_open(id)
}
fn can_recv_data(&mut self, id: StreamId) -> bool {
self.inner.can_recv_data(id)
fn is_recv_open(&mut self, id: StreamId) -> bool {
self.inner.is_recv_open(id)
}
}
/// Proxy.
impl<T: ControlPing> ControlPing for StreamSendClose<T> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)

43
src/proto/stream_send_open.rs
View File

@@ -3,7 +3,7 @@ use error::User::{InactiveStreamId, InvalidStreamId, StreamReset, Rejected, Unex
use frame::{Frame, SettingSet};
use proto::*;
///
/// Ensures that frames are sent on open streams in the appropriate state.
#[derive(Debug)]
pub struct StreamSendOpen<T> {
inner: T,
@@ -30,25 +30,7 @@ impl<T, U> StreamSendOpen<T>
}
}
/// Handles updates to `SETTINGS_MAX_CONCURRENT_STREAMS`.
///
/// > Indicates the maximum number of concurrent streams that the senderg will allow. This
/// > limit is directional: it applies to the number of streams that the sender permits
/// > the receiver to create. Initially, there is no limit to this value. It is
/// > recommended that this value be no smaller than 100, so as to not unnecessarily limit
/// > parallelism.
/// >
/// > A value of 0 for SETTINGS_MAX_CONCURRENT_STREAMS SHOULD NOT be treated as special by
/// > endpoints. A zero value does prevent the creation of new streams; however, this can
/// > also happen for any limit that is exhausted with active streams. Servers SHOULD only
/// > set a zero value for short durations; if a server does not wish to accept requests,
/// > closing the connection is more appropriate.
///
/// > An endpoint that wishes to reduce the value of SETTINGS_MAX_CONCURRENT_STREAMS to a
/// > value that is below the current number of open streams can either close streams that
/// > exceed the new value or allow streams to complete.
///
/// This module does NOT close streams when the setting changes.
/// Handles updates to `SETTINGS_MAX_CONCURRENT_STREAMS` from the remote peer.
impl<T: ApplySettings> ApplySettings for StreamSendOpen<T> {
fn apply_local_settings(&mut self, set: &SettingSet) -> Result<(), ConnectionError> {
self.inner.apply_local_settings(set)
@@ -61,6 +43,7 @@ impl<T: ApplySettings> ApplySettings for StreamSendOpen<T> {
}
}
/// Proxy.
impl<T> Stream for StreamSendOpen<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: ControlStreams,
@@ -74,6 +57,7 @@ impl<T> Stream for StreamSendOpen<T>
}
}
/// Helper.
impl<T: ControlStreams> StreamSendOpen<T> {
fn check_not_reset(&self, id: StreamId) -> Result<(), ConnectionError> {
// Ensure that the stream hasn't been closed otherwise.
@@ -84,6 +68,7 @@ impl<T: ControlStreams> StreamSendOpen<T> {
}
}
/// Ensures that frames are sent on open streams in the appropriate state.
impl<T, U> Sink for StreamSendOpen<T>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ControlStreams,
@@ -143,7 +128,7 @@ impl<T, U> Sink for StreamSendOpen<T>
// the stream is open (i.e. has already sent headers).
_ => {
self.check_not_reset(id)?;
if !self.inner.can_send_data(id) {
if !self.inner.is_send_open(id) {
return Err(InactiveStreamId.into());
}
}
@@ -157,6 +142,7 @@ impl<T, U> Sink for StreamSendOpen<T>
}
}
/// Proxy.
impl<T, U> ReadySink for StreamSendOpen<T>
where T: Stream<Item = Frame, Error = ConnectionError>,
T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
@@ -168,6 +154,7 @@ impl<T, U> ReadySink for StreamSendOpen<T>
}
}
/// Proxy.
impl<T: ControlStreams> ControlStreams for StreamSendOpen<T> {
fn local_valid_id(id: StreamId) -> bool {
T::local_valid_id(id)
@@ -229,11 +216,11 @@ impl<T: ControlStreams> ControlStreams for StreamSendOpen<T> {
self.inner.remote_active_len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_recv_window_size(old_sz, new_sz)
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
self.inner.update_inital_send_window_size(old_sz, new_sz)
}
@@ -245,15 +232,16 @@ impl<T: ControlStreams> ControlStreams for StreamSendOpen<T> {
self.inner.send_flow_controller(id)
}
fn can_send_data(&mut self, id: StreamId) -> bool {
self.inner.can_send_data(id)
fn is_send_open(&mut self, id: StreamId) -> bool {
self.inner.is_send_open(id)
}
fn can_recv_data(&mut self, id: StreamId) -> bool {
self.inner.can_recv_data(id)
fn is_recv_open(&mut self, id: StreamId) -> bool {
self.inner.is_recv_open(id)
}
}
/// Proxy.
impl<T: ControlFlow> ControlFlow for StreamSendOpen<T> {
fn poll_window_update(&mut self) -> Poll<WindowUpdate, ConnectionError> {
self.inner.poll_window_update()
@@ -264,6 +252,7 @@ impl<T: ControlFlow> ControlFlow for StreamSendOpen<T> {
}
}
/// Proxy.
impl<T: ControlPing> ControlPing for StreamSendOpen<T> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)

170
src/proto/stream_store.rs
View File

@@ -11,7 +11,12 @@ use std::marker::PhantomData;
/// Exposes stream states to "upper" layers of the transport (i.e. from StreamTracker up
/// to Connection).
pub trait ControlStreams {
/// Determines whether the given stream could theoretically be opened by the local
/// side of this connection.
fn local_valid_id(id: StreamId) -> bool;
/// Determines whether the given stream could theoretically be opened by the remote
/// side of this connection.
fn remote_valid_id(id: StreamId) -> bool;
/// Indicates whether this local endpoint may open streams (with HEADERS).
@@ -26,7 +31,7 @@ pub trait ControlStreams {
!Self::local_can_open()
}
/// Create a new stream in the OPEN state from the local side (i.e. as a Client).
/// Creates a new stream in the OPEN state from the local side (i.e. as a Client).
///
/// Must only be called when local_can_open returns true.
fn local_open(&mut self, id: StreamId, sz: WindowSize) -> Result<(), ConnectionError>;
@@ -46,19 +51,37 @@ pub trait ControlStreams {
/// Typically called when a server sends a response header.
fn remote_open_send_half(&mut self, id: StreamId, sz: WindowSize) -> Result<(), ConnectionError>;
// TODO push promise
// fn local_reserve(&mut self, id: StreamId) -> Result<(), ConnectionError>;
// fn remote_reserve(&mut self, id: StreamId) -> Result<(), ConnectionError>;
/// Close the local half of a stream so that the local side may not RECEIVE
/// Closes the send half of a stream.
///
/// Fails with a ProtocolError if send half of the stream was not open.
fn close_send_half(&mut self, id: StreamId) -> Result<(), ConnectionError>;
/// Closes the recv half of a stream.
///
/// Fails with a ProtocolError if recv half of the stream was not open.
fn close_recv_half(&mut self, id: StreamId) -> Result<(), ConnectionError>;
/// Resets the given stream.
///
/// If the stream was already reset, the stored cause is updated.
fn reset_stream(&mut self, id: StreamId, cause: Reason);
/// Get the reason the stream was reset, if it was reset.
fn get_reset(&self, id: StreamId) -> Option<Reason>;
/// Returns true if the given stream was opened by the local peer and is not yet
/// closed.
fn is_local_active(&self, id: StreamId) -> bool;
/// Returns true if the given stream was opened by the remote peer and is not yet
/// closed.
fn is_remote_active(&self, id: StreamId) -> bool;
/// Returns true if the given stream was opened and is not yet closed.
fn is_active(&self, id: StreamId) -> bool {
if Self::local_valid_id(id) {
self.is_local_active(id)
@@ -67,17 +90,31 @@ pub trait ControlStreams {
}
}
/// Returns the number of open streams initiated by the local peer.
fn local_active_len(&self) -> usize;
/// Returns the number of open streams initiated by the remote peer.
fn remote_active_len(&self) -> usize;
/// Returns true iff the recv half of the given stream is open.
fn is_recv_open(&mut self, id: StreamId) -> bool;
/// Returns true iff the send half of the given stream is open.
fn is_send_open(&mut self, id: StreamId) -> bool;
/// If the given stream ID is active and able to recv data, get its mutable recv flow
/// control state.
fn recv_flow_controller(&mut self, id: StreamId) -> Option<&mut FlowControlState>;
/// If the given stream ID is active and able to send data, get its mutable send flow
/// control state.
fn send_flow_controller(&mut self, id: StreamId) -> Option<&mut FlowControlState>;
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32);
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32);
/// Updates the initial window size for the local peer.
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize);
fn can_send_data(&mut self, id: StreamId) -> bool;
fn can_recv_data(&mut self, id: StreamId) -> bool;
/// Updates the initial window size for the remote peer.
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize);
}
/// Holds the underlying stream state to be accessed by upper layers.
@@ -115,41 +152,6 @@ impl<T, P, U> StreamStore<T, P>
}
}
impl<T, P> Stream for StreamStore<T, P>
where T: Stream<Item = Frame, Error = ConnectionError>,
{
type Item = Frame;
type Error = ConnectionError;
fn poll(&mut self) -> Poll<Option<Frame>, ConnectionError> {
self.inner.poll()
}
}
impl<T, P, U> Sink for StreamStore<T, P>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
{
type SinkItem = Frame<U>;
type SinkError = ConnectionError;
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Frame<U>, ConnectionError> {
self.inner.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), ConnectionError> {
self.inner.poll_complete()
}
}
impl<T, P, U> ReadySink for StreamStore<T, P>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
{
fn poll_ready(&mut self) -> Poll<(), ConnectionError> {
self.inner.poll_ready()
}
}
impl<T, P: Peer> StreamStore<T, P> {
pub fn get_active(&mut self, id: StreamId) -> Option<&StreamState> {
assert!(!id.is_zero());
@@ -192,21 +194,25 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
P::local_can_open()
}
/// Open a new stream from the local side (i.e. as a Client).
fn local_open(&mut self, id: StreamId, sz: WindowSize) -> Result<(), ConnectionError> {
assert!(Self::local_valid_id(id));
debug_assert!(Self::local_can_open());
debug_assert!(!self.local_active.contains_key(&id));
if !Self::local_valid_id(id) || !Self::local_can_open() {
return Err(ProtocolError.into());
}
if self.local_active.contains_key(&id) {
return Err(ProtocolError.into());
}
self.local_active.insert(id, StreamState::new_open_sending(sz));
Ok(())
}
/// Open a new stream from the remote side (i.e. as a Server).
fn remote_open(&mut self, id: StreamId, sz: WindowSize) -> Result<(), ConnectionError> {
assert!(Self::remote_valid_id(id));
debug_assert!(Self::remote_can_open());
debug_assert!(!self.remote_active.contains_key(&id));
if !Self::remote_valid_id(id) || !Self::remote_can_open() {
return Err(ProtocolError.into());
}
if self.remote_active.contains_key(&id) {
return Err(ProtocolError.into());
}
self.remote_active.insert(id, StreamState::new_open_recving(sz));
Ok(())
@@ -275,6 +281,20 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
self.remote_active.contains_key(&id)
}
fn is_send_open(&mut self, id: StreamId) -> bool {
match self.get_active(id) {
Some(s) => s.is_send_open(),
None => false,
}
}
fn is_recv_open(&mut self, id: StreamId) -> bool {
match self.get_active(id) {
Some(s) => s.is_recv_open(),
None => false,
}
}
fn local_active_len(&self) -> usize {
self.local_active.len()
}
@@ -283,7 +303,7 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
self.remote_active.len()
}
fn update_inital_recv_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_recv_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
if new_sz < old_sz {
let decr = old_sz - new_sz;
@@ -315,7 +335,7 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
}
}
fn update_inital_send_window_size(&mut self, old_sz: u32, new_sz: u32) {
fn update_inital_send_window_size(&mut self, old_sz: WindowSize, new_sz: WindowSize) {
if new_sz < old_sz {
let decr = old_sz - new_sz;
@@ -366,22 +386,47 @@ impl<T, P: Peer> ControlStreams for StreamStore<T, P> {
self.remote_active.get_mut(&id).and_then(|s| s.send_flow_controller())
}
}
}
fn can_send_data(&mut self, id: StreamId) -> bool {
match self.get_active(id) {
Some(s) => s.can_send_data(),
None => false,
}
}
/// Proxy.
impl<T, P> Stream for StreamStore<T, P>
where T: Stream<Item = Frame, Error = ConnectionError>,
{
type Item = Frame;
type Error = ConnectionError;
fn can_recv_data(&mut self, id: StreamId) -> bool {
match self.get_active(id) {
Some(s) => s.can_recv_data(),
None => false,
}
fn poll(&mut self) -> Poll<Option<Frame>, ConnectionError> {
self.inner.poll()
}
}
/// Proxy.
impl<T, P, U> Sink for StreamStore<T, P>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
{
type SinkItem = Frame<U>;
type SinkError = ConnectionError;
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Frame<U>, ConnectionError> {
self.inner.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), ConnectionError> {
self.inner.poll_complete()
}
}
/// Proxy.
impl<T, P, U> ReadySink for StreamStore<T, P>
where T: Sink<SinkItem = Frame<U>, SinkError = ConnectionError>,
T: ReadySink,
{
fn poll_ready(&mut self) -> Poll<(), ConnectionError> {
self.inner.poll_ready()
}
}
/// Proxy.
impl<T: ApplySettings, P> ApplySettings for StreamStore<T, P> {
fn apply_local_settings(&mut self, set: &frame::SettingSet) -> Result<(), ConnectionError> {
self.inner.apply_local_settings(set)
@@ -392,6 +437,7 @@ impl<T: ApplySettings, P> ApplySettings for StreamStore<T, P> {
}
}
/// Proxy.
impl<T: ControlPing, P> ControlPing for StreamStore<T, P> {
fn start_ping(&mut self, body: PingPayload) -> StartSend<PingPayload, ConnectionError> {
self.inner.start_ping(body)