diff --git a/src/client.rs b/src/client.rs index cd309bc..fd2ec71 100644 --- a/src/client.rs +++ b/src/client.rs @@ -177,6 +177,7 @@ use tokio_io::io::WriteAll; use std::fmt; use std::marker::PhantomData; use std::time::Duration; +use std::usize; /// Performs the HTTP/2.0 connection handshake. /// @@ -344,6 +345,12 @@ pub struct Builder { /// Time to keep locally reset streams around before reaping. reset_stream_duration: Duration, + /// Initial maximum number of locally initiated (send) streams. + /// After receiving a Settings frame from the remote peer, + /// the connection will overwrite this value with the + /// MAX_CONCURRENT_STREAMS specified in the frame. + initial_max_send_streams: usize, + /// Maximum number of locally reset streams to keep at a time. reset_stream_max: usize, @@ -671,6 +678,7 @@ impl Builder { Builder { reset_stream_duration: Duration::from_secs(proto::DEFAULT_RESET_STREAM_SECS), reset_stream_max: proto::DEFAULT_RESET_STREAM_MAX, + initial_max_send_streams: usize::MAX, settings: Default::default(), stream_id: 1.into(), } @@ -838,6 +846,48 @@ impl Builder { self } + /// Sets the initial maximum of locally initiated (send) streams. + /// + /// The initial settings will be overwritten by the remote peer when + /// the Settings frame is received. The new value will be set to the + /// `max_concurrent_streams()` from the frame. + /// + /// This setting prevents the caller from exceeding this number of + /// streams that are counted towards the concurrency limit. + /// + /// Sending streams past the limit returned by the peer will be treated + /// as a stream error of type PROTOCOL_ERROR or REFUSED_STREAM. + /// + /// See [Section 5.1.2] in the HTTP/2.0 spec for more details. + /// + /// [Section 5.1.2]: https://http2.github.io/http2-spec/#rfc.section.5.1.2 + /// + /// # Examples + /// + /// ``` + /// # extern crate h2; + /// # extern crate tokio_io; + /// # use tokio_io::*; + /// # use h2::client::*; + /// # + /// # fn doc(my_io: T) + /// # -> Handshake + /// # { + /// // `client_fut` is a future representing the completion of the HTTP/2.0 + /// // handshake. + /// let client_fut = Builder::new() + /// .initial_max_send_streams(1000) + /// .handshake(my_io); + /// # client_fut + /// # } + /// # + /// # pub fn main() {} + /// ``` + pub fn initial_max_send_streams(&mut self, initial: usize) -> &mut Self { + self.initial_max_send_streams = initial; + self + } + /// Sets the maximum number of concurrent locally reset streams. /// /// When a stream is explicitly reset by either calling @@ -1218,6 +1268,7 @@ where let connection = proto::Connection::new(codec, proto::Config { next_stream_id: self.builder.stream_id, + initial_max_send_streams: self.builder.initial_max_send_streams, reset_stream_duration: self.builder.reset_stream_duration, reset_stream_max: self.builder.reset_stream_max, settings: self.builder.settings.clone(), diff --git a/src/proto/connection.rs b/src/proto/connection.rs index 7f23794..c6d241e 100644 --- a/src/proto/connection.rs +++ b/src/proto/connection.rs @@ -46,6 +46,7 @@ where #[derive(Debug, Clone)] pub(crate) struct Config { pub next_stream_id: StreamId, + pub initial_max_send_streams: usize, pub reset_stream_duration: Duration, pub reset_stream_max: usize, pub settings: frame::Settings, @@ -80,7 +81,7 @@ where local_init_window_sz: config.settings .initial_window_size() .unwrap_or(DEFAULT_INITIAL_WINDOW_SIZE), - local_max_initiated: None, + initial_max_send_streams: config.initial_max_send_streams, local_next_stream_id: config.next_stream_id, local_push_enabled: config.settings.is_push_enabled(), local_reset_duration: config.reset_stream_duration, diff --git a/src/proto/streams/counts.rs b/src/proto/streams/counts.rs index 0e82f15..2d54279 100644 --- a/src/proto/streams/counts.rs +++ b/src/proto/streams/counts.rs @@ -32,7 +32,7 @@ impl Counts { pub fn new(peer: peer::Dyn, config: &Config) -> Self { Counts { peer, - max_send_streams: config.local_max_initiated.unwrap_or(usize::MAX), + max_send_streams: config.initial_max_send_streams, num_send_streams: 0, max_recv_streams: config.remote_max_initiated.unwrap_or(usize::MAX), num_recv_streams: 0, diff --git a/src/proto/streams/mod.rs b/src/proto/streams/mod.rs index 13ce2eb..7a95e67 100644 --- a/src/proto/streams/mod.rs +++ b/src/proto/streams/mod.rs @@ -26,17 +26,20 @@ use self::stream::Stream; use frame::{StreamId, StreamIdOverflow}; use proto::*; -use std::time::Duration; use bytes::Bytes; use http::{Request, Response}; +use std::time::Duration; #[derive(Debug)] pub struct Config { /// Initial window size of locally initiated streams pub local_init_window_sz: WindowSize, - /// Maximum number of locally initiated streams - pub local_max_initiated: Option, + /// Initial maximum number of locally initiated streams. + /// After receiving a Settings frame from the remote peer, + /// the connection will overwrite this value with the + /// MAX_CONCURRENT_STREAMS specified in the frame. + pub initial_max_send_streams: usize, /// The stream ID to start the next local stream with pub local_next_stream_id: StreamId, diff --git a/src/server.rs b/src/server.rs index 3a4c716..e1be793 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1026,6 +1026,8 @@ impl Future for Handshake let server = poll?.map(|codec| { let connection = proto::Connection::new(codec, Config { next_stream_id: 2.into(), + // Server does not need to locally initiate any streams + initial_max_send_streams: 0, reset_stream_duration: self.builder.reset_stream_duration, reset_stream_max: self.builder.reset_stream_max, settings: self.builder.settings.clone(),