Add initial_max_send_streams() as builder option (#242)

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<T: AsyncRead + AsyncWrite>(my_io: T)
+    /// # -> Handshake<T>
+    /// # {
+    /// // `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(),

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,

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,

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
+    /// Initial maximum number of locally initiated streams.
-    pub local_max_initiated: Option<usize>,
+    /// 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,

src/server.rs

@@ -1026,6 +1026,8 @@ impl<T, B: IntoBuf> Future for Handshake<T, B>
         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(),