From 2f7548859332663ae762e1f88e2482338780284b Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 28 Mar 2016 06:23:57 +0000 Subject: [PATCH 01/24] WindowSize: Implement Into and add `can_accept` method --- src/http/mod.rs | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/http/mod.rs b/src/http/mod.rs index 1fed8257..a64a30cf 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -522,6 +522,7 @@ impl WindowSize { pub fn new(size: i32) -> WindowSize { WindowSize(size) } + /// Returns the current size of the window. /// /// The size is actually allowed to become negative (for instance if the peer changes its @@ -529,8 +530,31 @@ impl WindowSize { pub fn size(&self) -> i32 { self.0 } + + /// Checks whether the window is large enough to accept the given number of bytes. + /// If this method returns `true`, it is guaranteed that `try_decrease` must succeed. + /// + /// ```rust + /// use solicit::http::WindowSize; + /// + /// let window = WindowSize::new(1000); + /// assert!(window.can_accept(5)); + /// assert!(window.can_accept(1000)); + /// assert!(!window.can_accept(1001)); + /// ``` + pub fn can_accept(&self, size: i32) -> bool { + self.0 >= size + } +} + +impl Into for WindowSize { + fn into(self) -> i32 { + self.size() + } } +pub const DEFAULT_MAX_WINDOW_SIZE: WindowSize = WindowSize(0xffff); + /// An enum representing the two possible HTTP schemes. #[derive(Debug, Copy, Clone, PartialEq)] pub enum HttpScheme { From 568a06f846e1bfa808b5cb15a9df7d36d57d4be1 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Sun, 17 Apr 2016 19:01:57 -0700 Subject: [PATCH 02/24] WindowSize: Implement PartialEq This allows `WindowSize`s to be compared for equality against `i32`s transparently... --- src/http/mod.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/http/mod.rs b/src/http/mod.rs index a64a30cf..e55995f7 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -553,6 +553,12 @@ impl Into for WindowSize { } } +impl PartialEq for WindowSize { + fn eq(&self, rhs: &i32) -> bool { + self.size() == *rhs + } +} + pub const DEFAULT_MAX_WINDOW_SIZE: WindowSize = WindowSize(0xffff); /// An enum representing the two possible HTTP schemes. From c333681727d345b21650324098ff5ee1e9829fa5 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Sun, 17 Apr 2016 19:05:35 -0700 Subject: [PATCH 03/24] HttpConnection: Return WindowSize structs rather than i32 from window size methods There is no point in coercing the internal `WindowSize` struct into an `i32` before returning it; in fact, this can easily be done by clients if they have such a requirement. --- src/http/connection.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/http/connection.rs b/src/http/connection.rs index 18fcfdcf..d47a3f2b 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -274,7 +274,7 @@ impl<'a, S> HttpConnectionSender<'a, S> } // Adjust the flow control window... try!(self.conn.decrease_out_window(frame.payload_len())); - trace!("New OUT WINDOW size = {}", self.conn.out_window_size()); + trace!("New OUT WINDOW size = {:?}", self.conn.out_window_size()); // ...and now send it out. self.send_frame(frame) } @@ -352,13 +352,14 @@ impl HttpConnection { /// Returns the current size of the inbound flow control window (i.e. the number of octets that /// the connection will accept and the peer will send at most, unless the window is updated). - pub fn in_window_size(&self) -> i32 { - self.in_window_size.size() + pub fn in_window_size(&self) -> WindowSize { + self.in_window_size } + /// Returns the current size of the outbound flow control window (i.e. the number of octets /// that can be sent on the connection to the peer without violating flow control). - pub fn out_window_size(&self) -> i32 { - self.out_window_size.size() + pub fn out_window_size(&self) -> WindowSize { + self.out_window_size } /// The method processes the next frame provided by the given `ReceiveFrame` instance, expecting @@ -472,7 +473,7 @@ impl HttpConnection { session: &mut Sess) -> HttpResult<()> { try!(self.decrease_in_window(frame.payload_len())); - trace!("New IN WINDOW size = {}", self.in_window_size()); + trace!("New IN WINDOW size = {:?}", self.in_window_size()); try!(session.new_data_chunk(frame.get_stream_id(), &frame.data, self)); // TODO(mlalic): Should the connection separately signal the decrease in the flow control // window? For now, it is expected that the data callback is enough, as the From 7764212ea0e380888c09e0ae229ca1a26a0a49fc Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 26 Oct 2015 20:10:53 +0100 Subject: [PATCH 04/24] HttpConnection: Add methods for sending window updates --- src/http/connection.rs | 75 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 74 insertions(+), 1 deletion(-) diff --git a/src/http/connection.rs b/src/http/connection.rs index d47a3f2b..908d55b5 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -279,6 +279,29 @@ impl<'a, S> HttpConnectionSender<'a, S> self.send_frame(frame) } + /// Sends a window update frame for the peer's connection level flow control window. + pub fn send_connection_window_update(&mut self, increment: u32) -> HttpResult<()> { + if increment == 0 { + warn!("Tried to increase window by zero, which would be invalid; frame not sent."); + return Ok(()); + } + let frame = WindowUpdateFrame::for_connection(increment); + self.send_frame(frame) + } + + /// Sends a window update frame for the given stream's flow control window. + pub fn send_stream_window_update(&mut self, + stream_id: StreamId, + increment: u32) + -> HttpResult<()> { + if increment == 0 { + warn!("Tried to increase window by zero, which would be invalid; frame not sent."); + return Ok(()); + } + let frame = WindowUpdateFrame::for_stream(stream_id, increment); + self.send_frame(frame) + } + /// Sends the chunk of data provided by the given `DataPrioritizer`. /// /// # Returns @@ -575,7 +598,7 @@ mod tests { use http::tests::common::{build_mock_http_conn, StubDataPrioritizer, TestSession, MockReceiveFrame, MockSendFrame}; use http::frame::{Frame, DataFrame, HeadersFrame, RstStreamFrame, GoawayFrame, SettingsFrame, - PingFrame, pack_header, RawFrame, FrameIR}; + WindowUpdateFrame, PingFrame, pack_header, RawFrame, FrameIR}; use http::{HttpResult, HttpScheme, Header, OwnedHeader, ErrorCode}; use hpack; @@ -891,6 +914,56 @@ mod tests { expect_frame_list(expected, sender.sent); } + #[test] + fn test_send_connection_window_update_non_zero() { + let increment = 100; + let expected = vec![ + HttpFrame::WindowUpdateFrame(WindowUpdateFrame::for_connection(increment)), + ]; + + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + conn.sender(&mut sender).send_connection_window_update(increment).unwrap(); + + expect_frame_list(expected, sender.sent); + } + + #[test] + fn test_send_connection_window_update_is_zero() { + let increment = 0; + + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + conn.sender(&mut sender).send_connection_window_update(increment).unwrap(); + + expect_frame_list(vec![], sender.sent); + } + + #[test] + fn test_send_stream_window_update_non_zero() { + let increment = 100; + let expected = vec![ + HttpFrame::WindowUpdateFrame(WindowUpdateFrame::for_stream(1, increment)), + ]; + + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + conn.sender(&mut sender).send_stream_window_update(1, increment).unwrap(); + + expect_frame_list(expected, sender.sent); + } + + #[test] + fn test_send_stream_window_update_is_zero() { + let increment = 0; + + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + conn.sender(&mut sender).send_stream_window_update(1, increment).unwrap(); + + expect_frame_list(vec![], sender.sent); + } + /// Tests that the `HttpConnection` correctly notifies the session on a /// new headers frame, with no continuation. #[test] From 5c0459f53be35cb57852a0078ac9100917a199e0 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 26 Oct 2015 20:22:04 +0100 Subject: [PATCH 05/24] HttpConnection: Add method for increasing inbound window size --- src/http/connection.rs | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/src/http/connection.rs b/src/http/connection.rs index 908d55b5..78db88ca 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -385,6 +385,15 @@ impl HttpConnection { self.out_window_size } + /// Increases the size of the inbound connection flow control window by the given delta. + /// + /// If this would cause the window to overflow the maximum value of 2^31 - 1, returns an error. + /// The method **does not** automatically send any window update frames. It is the caller's + /// responsibility to make sure that the peer is notified of the window increase. + pub fn increase_connection_window_size(&mut self, delta: u32) -> HttpResult<()> { + self.in_window_size.try_increase(delta).map_err(|_| HttpError::WindowSizeOverflow) + } + /// The method processes the next frame provided by the given `ReceiveFrame` instance, expecting /// it to be a SETTINGS frame. /// Additionally, the frame cannot be an ACK settings frame, but rather it should contain the @@ -599,7 +608,7 @@ mod tests { MockReceiveFrame, MockSendFrame}; use http::frame::{Frame, DataFrame, HeadersFrame, RstStreamFrame, GoawayFrame, SettingsFrame, WindowUpdateFrame, PingFrame, pack_header, RawFrame, FrameIR}; - use http::{HttpResult, HttpScheme, Header, OwnedHeader, ErrorCode}; + use http::{HttpResult, HttpError, HttpScheme, Header, OwnedHeader, ErrorCode}; use hpack; /// A helper function that performs a `send_frame` operation on the given @@ -964,6 +973,21 @@ mod tests { expect_frame_list(vec![], sender.sent); } + #[test] + fn test_increase_connection_window() { + let mut conn = build_mock_http_conn(); + assert_eq!(conn.in_window_size(), 0xffff); + conn.increase_connection_window_size(500).unwrap(); + assert_eq!(conn.in_window_size(), 0xffff + 500); + conn.increase_connection_window_size(5).unwrap(); + assert_eq!(conn.in_window_size(), 0xffff + 500 + 5); + // Overflow! + assert!(match conn.increase_connection_window_size(0x7fffffff).err().unwrap() { + HttpError::WindowSizeOverflow => true, + _ => false, + }); + } + /// Tests that the `HttpConnection` correctly notifies the session on a /// new headers frame, with no continuation. #[test] From 8c6297bb0da2cee13c6cef9d744595b065d1e33d Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 26 Oct 2015 21:30:33 +0100 Subject: [PATCH 06/24] HttpConnection: Notify the session according to detected window updates This commit adds two methods to the `Session` trait that allow it to be notified when a new WINDOW_UPDATE frame signals an increase in the outbound window size for both the connection-level, as well as the stream-level windows. --- src/http/connection.rs | 67 ++++++++++++++++++++++++++++++++++++++-- src/http/session.rs | 24 ++++++++++++++ src/http/tests/common.rs | 24 +++++++++++++- 3 files changed, 111 insertions(+), 4 deletions(-) diff --git a/src/http/connection.rs b/src/http/connection.rs index 78db88ca..5adc3617 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -485,9 +485,9 @@ impl HttpConnection { frame.debug_data(), self) } - HttpFrame::WindowUpdateFrame(_) => { + HttpFrame::WindowUpdateFrame(frame) => { debug!("WINDOW_UPDATE frame received"); - Ok(()) + self.handle_window_update(frame, session) } HttpFrame::UnknownFrame(frame) => { debug!("Unknown frame received; raw = {:?}", frame); @@ -565,7 +565,6 @@ impl HttpConnection { -> HttpResult<()> { if !frame.is_ack() { // TODO: Actually handle the settings change before sending out the ACK - // sending out the ACK. trace!("New settings frame {:#?}", frame); try!(session.new_settings(frame.settings, self)); } @@ -573,6 +572,26 @@ impl HttpConnection { Ok(()) } + /// Private helper method that handles an incoming `WindowUpdateFrame`. + fn handle_window_update(&mut self, + frame: WindowUpdateFrame, + session: &mut Sess) + -> HttpResult<()> { + if frame.get_stream_id() == 0 { + // TODO: If overflow does occur, notify the session so that it can try to send a + // GOAWAY before tearing down the connection. + try!(self.out_window_size.try_increase(frame.increment()) + .map_err(|_| HttpError::WindowSizeOverflow)); + try!(session.on_connection_out_window_update(self)); + } else { + try!(session.on_stream_out_window_update(frame.get_stream_id(), + frame.increment(), + self)); + } + + Ok(()) + } + /// Internal helper method that decreases the outbound flow control window size. fn decrease_out_window(&mut self, size: u32) -> HttpResult<()> { // The size by which we decrease the window must be at most 2^31 - 1. We should be able to @@ -1098,6 +1117,48 @@ mod tests { assert_eq!(session.rst_streams.len(), 0); } + #[test] + fn test_conn_window_update() { + let frames = vec![ + HttpFrame::WindowUpdateFrame(WindowUpdateFrame::for_connection(100)), + ]; + let mut conn = HttpConnection::new(HttpScheme::Http); + let mut session = TestSession::new(); + let mut frame_provider = MockReceiveFrame::new(frames); + + conn.handle_next_frame(&mut frame_provider, &mut session).unwrap(); + + assert_eq!(session.conn_window_updates.len(), 1); + assert_eq!(session.conn_window_updates[0], 0xffff + 100); + // Nothing happened with the rest... + assert_eq!(session.stream_window_updates.len(), 0); + assert_eq!(session.goaways.len(), 0); + assert_eq!(session.curr_header, 0); + assert_eq!(session.curr_chunk, 0); + assert_eq!(session.rst_streams.len(), 0); + } + + #[test] + fn test_conn_stream_window_update() { + let frames = vec![ + HttpFrame::WindowUpdateFrame(WindowUpdateFrame::for_stream(1, 100)), + ]; + let mut conn = HttpConnection::new(HttpScheme::Http); + let mut session = TestSession::new(); + let mut frame_provider = MockReceiveFrame::new(frames); + + conn.handle_next_frame(&mut frame_provider, &mut session).unwrap(); + + assert_eq!(session.stream_window_updates.len(), 1); + assert_eq!(session.stream_window_updates[0], (1, 100)); + // Nothing happened with the rest... + assert_eq!(session.conn_window_updates.len(), 0); + assert_eq!(session.goaways.len(), 0); + assert_eq!(session.curr_header, 0); + assert_eq!(session.curr_chunk, 0); + assert_eq!(session.rst_streams.len(), 0); + } + /// Tests that the connection flow control windows have the correct size when the /// HttpConnection is just created. #[test] diff --git a/src/http/session.rs b/src/http/session.rs index 0686b4cd..cc9a416d 100644 --- a/src/http/session.rs +++ b/src/http/session.rs @@ -78,6 +78,30 @@ pub trait Session { debug_data: debug_data.map(|data| data.to_vec()), })) } + + /// Notifies the `Session` that the connection's outbound flow control window was updated. + /// + /// The default implementation of the method ignores any change. + /// + /// Concrete implementations can rely on this to, for example, trigger more writes on a + /// connection that was previously blocked on flow control (rather than on socket IO). + fn on_connection_out_window_update(&mut self, _conn: &mut HttpConnection) -> HttpResult<()> { + Ok(()) + } + + /// Notifies the `Session` that the given stream's outbound flow control window should be + /// updated. Unlike the `on_connection_out_window_update`, the new size is not provided, but + /// rather the size of the increment. This is due to the fact that the `HttpConnection` does + /// not handle individual streams, but expects the session layer to be in charge of that. + /// + /// The default implementation of the method ignores any change. + fn on_stream_out_window_update(&mut self, + _stream_id: StreamId, + _increment: u32, + _conn: &mut HttpConnection) + -> HttpResult<()> { + Ok(()) + } } /// A newtype for an iterator over `Stream`s saved in a `SessionState`. diff --git a/src/http/tests/common.rs b/src/http/tests/common.rs index 20952257..6b4b06e7 100644 --- a/src/http/tests/common.rs +++ b/src/http/tests/common.rs @@ -6,7 +6,7 @@ use std::cell::{RefCell, Cell}; use std::borrow::Cow; use std::io::{Cursor, Read, Write}; -use http::{HttpResult, HttpScheme, StreamId, Header, OwnedHeader, ErrorCode}; +use http::{HttpResult, HttpScheme, StreamId, Header, OwnedHeader, ErrorCode, WindowSize}; use http::frame::{RawFrame, FrameIR, FrameHeader, pack_header, HttpSetting, PingFrame}; use http::session::{Session, DefaultSessionState, SessionState, Stream, StreamState, StreamDataChunk, StreamDataError}; @@ -252,6 +252,10 @@ pub struct TestSession { pub pings: Vec, /// All the ping ack data received pub pongs: Vec, + /// All connection window updates as the value of the new connection window size + pub conn_window_updates: Vec, + /// All stream window updates as a pair of the stream id and the increment value. + pub stream_window_updates: Vec<(StreamId, u32)>, } impl TestSession { @@ -268,6 +272,8 @@ impl TestSession { goaways: Vec::new(), pings: Vec::new(), pongs: Vec::new(), + conn_window_updates: Vec::new(), + stream_window_updates: Vec::new(), } } @@ -285,6 +291,8 @@ impl TestSession { goaways: Vec::new(), pings: Vec::new(), pongs: Vec::new(), + conn_window_updates: Vec::new(), + stream_window_updates: Vec::new(), } } } @@ -353,6 +361,20 @@ impl Session for TestSession { self.pongs.push(ping.opaque_data()); Ok(()) } + + fn on_connection_out_window_update(&mut self, conn: &mut HttpConnection) -> HttpResult<()> { + self.conn_window_updates.push(conn.out_window_size()); + Ok(()) + } + + fn on_stream_out_window_update(&mut self, + stream_id: StreamId, + increment: u32, + _: &mut HttpConnection) + -> HttpResult<()> { + self.stream_window_updates.push((stream_id, increment)); + Ok(()) + } } /// A stream that can be used for testing purposes. From 7b89260d622cec2ec2f4541ad08c7a8c51039bda Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Tue, 27 Oct 2015 14:15:21 +0100 Subject: [PATCH 07/24] HttpConnection: Notify session on window size decreases Two new methods are added to the `Session` trait that allow it to receive notice when either the connection-level or a stream-level inbound window size decreases. These calls will, as a rule, always be emitted in pairs by the `HttpConnection` upon parsing a flow-control subject frame (at the moment this is only the DATA frame, as per the HTTP/2 spec). --- src/http/connection.rs | 36 ++++++++++++++++++++++++++---------- src/http/session.rs | 17 +++++++++++++++++ src/http/tests/common.rs | 23 +++++++++++++++++++++++ 3 files changed, 66 insertions(+), 10 deletions(-) diff --git a/src/http/connection.rs b/src/http/connection.rs index 5adc3617..4636366c 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -504,13 +504,19 @@ impl HttpConnection { frame: DataFrame, session: &mut Sess) -> HttpResult<()> { + // Decrease the connection inbound window... try!(self.decrease_in_window(frame.payload_len())); trace!("New IN WINDOW size = {:?}", self.in_window_size()); + try!(session.on_connection_in_window_decrease(self)); + + // ...as well as the stream's... + try!(session.on_stream_in_window_decrease( + frame.get_stream_id(), + frame.payload_len(), + self)); + + // ...before passing off the actual data chunk to the session. try!(session.new_data_chunk(frame.get_stream_id(), &frame.data, self)); - // TODO(mlalic): Should the connection separately signal the decrease in the flow control - // window? For now, it is expected that the data callback is enough, as the - // session would be able to inspect the new window size there (and know that - // it was affected by the data already). if frame.is_set(DataFlag::EndStream) { debug!("End of stream {}", frame.get_stream_id()); @@ -1041,13 +1047,21 @@ mod tests { conn.handle_next_frame(&mut frame_provider, &mut session).unwrap(); - // A poor man's mock... // The header callback was not called assert_eq!(session.curr_header, 0); // and exactly one chunk seen. assert_eq!(session.curr_chunk, 1); + // which caused the stream's window to decrease + assert_eq!(session.stream_window_decreases.len(), 1); + assert_eq!(session.stream_window_decreases[0], (1, 6)); + + // as well as the connection's... + let new_conn_in_window = 65_535 - 6; + assert_eq!(conn.in_window_size(), new_conn_in_window); + // ...which was reported to the session + assert_eq!(session.conn_in_window_decreases, vec![new_conn_in_window]); + assert_eq!(session.rst_streams.len(), 0); - assert_eq!(conn.in_window_size(), 65_535 - 6); } /// Tests that the session gets the correct values for the headers and data @@ -1060,10 +1074,8 @@ mod tests { hpack::Encoder::new().encode( expected_headers.iter().map(|h| (&h.0[..], &h.1[..]))), 1)), - HttpFrame::DataFrame(DataFrame::new(1)), { - let frame = DataFrame::with_data(1, &b"1234"[..]); - HttpFrame::DataFrame(frame) - }, + HttpFrame::DataFrame(DataFrame::new(1)), + HttpFrame::DataFrame(DataFrame::with_data(1, &b"1234"[..])), ]; let mut conn = HttpConnection::new(HttpScheme::Http); let mut session = TestSession::new_verify(vec![expected_headers], @@ -1074,6 +1086,10 @@ mod tests { conn.handle_next_frame(&mut frame_provider, &mut session).unwrap(); conn.handle_next_frame(&mut frame_provider, &mut session).unwrap(); assert_eq!(conn.in_window_size(), 65_535 - 4); + assert_eq!(session.conn_in_window_decreases, vec![65_535, 65_535 - 4]); + assert_eq!(session.stream_window_decreases.len(), 2); + assert_eq!(session.stream_window_decreases[0], (1, 0)); + assert_eq!(session.stream_window_decreases[1], (1, 4)); // Two chunks and one header processed? assert_eq!(session.curr_chunk, 2); diff --git a/src/http/session.rs b/src/http/session.rs index cc9a416d..0af09579 100644 --- a/src/http/session.rs +++ b/src/http/session.rs @@ -102,6 +102,23 @@ pub trait Session { -> HttpResult<()> { Ok(()) } + + /// Notifies the `Session` that the connection-level inbound flow control window has decreased. + /// The new value can be obtained from the given `HttpConnection` instance. + fn on_connection_in_window_decrease(&mut self, _conn: &mut HttpConnection) -> HttpResult<()> { + Ok(()) + } + + /// Notifies the `Session` that the given stream's inbound flow control window has decreased by + /// the given number of octets. + fn on_stream_in_window_decrease( + &mut self, + _stream_id: StreamId, + _size: u32, + _conn: &mut HttpConnection) + -> HttpResult<()> { + Ok(()) + } } /// A newtype for an iterator over `Stream`s saved in a `SessionState`. diff --git a/src/http/tests/common.rs b/src/http/tests/common.rs index 6b4b06e7..758cb282 100644 --- a/src/http/tests/common.rs +++ b/src/http/tests/common.rs @@ -256,6 +256,10 @@ pub struct TestSession { pub conn_window_updates: Vec, /// All stream window updates as a pair of the stream id and the increment value. pub stream_window_updates: Vec<(StreamId, u32)>, + /// All connection window updates as the new size of the connection window after the update + pub conn_in_window_decreases: Vec, + /// All stream window decreases that the session was notified of. + pub stream_window_decreases: Vec<(StreamId, u32)>, } impl TestSession { @@ -274,6 +278,8 @@ impl TestSession { pongs: Vec::new(), conn_window_updates: Vec::new(), stream_window_updates: Vec::new(), + conn_in_window_decreases: Vec::new(), + stream_window_decreases: Vec::new(), } } @@ -293,6 +299,8 @@ impl TestSession { pongs: Vec::new(), conn_window_updates: Vec::new(), stream_window_updates: Vec::new(), + conn_in_window_decreases: Vec::new(), + stream_window_decreases: Vec::new(), } } } @@ -375,6 +383,21 @@ impl Session for TestSession { self.stream_window_updates.push((stream_id, increment)); Ok(()) } + + fn on_connection_in_window_decrease(&mut self, conn: &mut HttpConnection) -> HttpResult<()> { + self.conn_in_window_decreases.push(conn.in_window_size()); + Ok(()) + } + + fn on_stream_in_window_decrease( + &mut self, + stream_id: StreamId, + size: u32, + _: &mut HttpConnection) + -> HttpResult<()> { + self.stream_window_decreases.push((stream_id, size)); + Ok(()) + } } /// A stream that can be used for testing purposes. From f998f4a1d9edbce88600768e8c9259d9ffb08a90 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Tue, 27 Oct 2015 14:44:46 +0100 Subject: [PATCH 08/24] SessionState: Manage state Entries, not naked `Stream`s The state now has an entry per stream, instead of storing just the raw stream. This is in preparation for extending the state to be able to track more stream-related information, that does not belong on the Stream trait itself. (The Stream trait is supposed to be the bridge between the session and its client that is interested only in stream-level events; requiring those clients to also implement tracking various book-keeping stream state is not only redundant---all streams will need to track the same state in the same way---but also not in line with what the Stream trait is supposed to represent.) --- src/client/async.rs | 2 +- src/http/priority.rs | 5 ++- src/http/session.rs | 75 ++++++++++++++++++++++++++++++-------------- src/server/mod.rs | 7 ++--- 4 files changed, 60 insertions(+), 29 deletions(-) diff --git a/src/client/async.rs b/src/client/async.rs index e8441b18..0dfa04d3 100644 --- a/src/client/async.rs +++ b/src/client/async.rs @@ -563,7 +563,7 @@ impl ClientService { /// stream might end up being closed by the server with an error. fn handle_closed(&mut self) { let done = self.conn.state.get_closed(); - for stream in done { + for stream in done.into_iter().map(|e| e.stream()) { self.send_response(stream); self.outstanding_reqs -= 1; } diff --git a/src/http/priority.rs b/src/http/priority.rs index c4c54512..88fdf022 100644 --- a/src/http/priority.rs +++ b/src/http/priority.rs @@ -48,7 +48,10 @@ impl<'a, 'b, State> DataPrioritizer for SimplePrioritizer<'a, 'b, State> { fn get_next_chunk(&mut self) -> HttpResult> { // Returns the data of the first stream that has data to be written. - for (stream_id, stream) in self.state.iter().filter(|&(_, ref s)| !s.is_closed_local()) { + let streams = self.state.iter() + .map(|(id, e)| (id, e.stream_mut())) + .filter(|&(_, ref s)| !s.is_closed_local()); + for (stream_id, stream) in streams { let res = stream.get_data_chunk(self.buf); match res { Ok(StreamDataChunk::Last(total)) => { diff --git a/src/http/session.rs b/src/http/session.rs index 0af09579..4e355ea6 100644 --- a/src/http/session.rs +++ b/src/http/session.rs @@ -125,19 +125,40 @@ pub trait Session { /// /// Allows `SessionState` implementations to return iterators over its session without being forced /// to declare them as associated types. -pub struct StreamIter<'a, S: Stream + 'a>(Box + 'a>); +pub struct StreamIter<'a, S: Stream + 'a>(Box)> + 'a>); impl<'a, S> Iterator for StreamIter<'a, S> where S: Stream + 'a { - type Item = (&'a StreamId, &'a mut S); + type Item = (&'a StreamId, &'a mut Entry); #[inline] - fn next(&mut self) -> Option<(&'a StreamId, &'a mut S)> { + fn next(&mut self) -> Option<(&'a StreamId, &'a mut Entry)> { self.0.next() } } +/// A struct representing a single entry in the `SessionState`. The `Entry` represents all relevant +/// information for a single HTTP/2 stream. +pub struct Entry where S: Stream { + stream: S, +} + +impl Entry where S: Stream { + /// Create a new `Entry` with the given `Stream` + pub fn new(stream: S) -> Entry { + Entry { + stream: stream, + } + } + /// Consumes the `Entry`, returning the underlying `Stream` instance + pub fn stream(self) -> S { self.stream } + /// Returns a reference to the `Stream` + pub fn stream_ref(&self) -> &S { &self.stream } + /// Returns a mutable reference to the `Stream` + pub fn stream_mut(&mut self) -> &mut S { &mut self.stream } +} + /// A trait defining a set of methods for accessing and influencing an HTTP/2 session's state. /// /// This trait is tightly coupled to a `Stream`-based session layer implementation. Particular @@ -159,15 +180,13 @@ pub trait SessionState { /// stream. /// TODO(mlalic): Allow the exact error to propagate out. fn insert_incoming(&mut self, id: StreamId, stream: Self::Stream) -> Result<(), ()>; - /// Returns a reference to a `Stream` with the given `StreamId`, if it is found in the current - /// session. - fn get_stream_ref(&self, stream_id: StreamId) -> Option<&Self::Stream>; - /// Returns a mutable reference to a `Stream` with the given `StreamId`, if it is found in the - /// current session. - fn get_stream_mut(&mut self, stream_id: StreamId) -> Option<&mut Self::Stream>; + /// Returns a reference to the `Entry` for the stream with the given id. + fn get_entry_ref(&self, id: StreamId) -> Option<&Entry>; + /// Returns a mutable reference to the `Entry` for the stream with the given id. + fn get_entry_mut(&mut self, id: StreamId) -> Option<&mut Entry>; /// Removes the stream with the given `StreamId` from the session. If the stream was found in /// the session, it is returned in the result. - fn remove_stream(&mut self, stream_id: StreamId) -> Option; + fn remove_stream(&mut self, stream_id: StreamId) -> Option>; /// Returns an iterator over the streams currently found in the session. fn iter(&mut self) -> StreamIter; @@ -175,22 +194,33 @@ pub trait SessionState { /// The number of streams tracked by this state object fn len(&self) -> usize; + /// Returns a reference to a `Stream` with the given `StreamId`, if it is found in the current + /// session. + fn get_stream_ref(&self, stream_id: StreamId) -> Option<&Self::Stream> { + self.get_entry_ref(stream_id).map(|e| e.stream_ref()) + } + + /// Returns a mutable reference to a `Stream` with the given `StreamId`, if it is found in the + /// current session. + fn get_stream_mut(&mut self, stream_id: StreamId) -> Option<&mut Self::Stream> { + self.get_entry_mut(stream_id).map(|e| e.stream_mut()) + } + /// Returns all streams that are closed and tracked by the session state. /// /// The streams are moved out of the session state. /// /// The default implementations relies on the `iter` implementation to find the closed streams /// first and then calls `remove_stream` on all of them. - fn get_closed(&mut self) -> Vec { + fn get_closed(&mut self) -> Vec> { let ids: Vec = self.iter() - .filter_map(|(id, s)| { - if s.is_closed() { + .filter_map(|(id, e)| { + if e.stream_ref().is_closed() { Some(*id) } else { None } - }) - .collect(); + }).collect(); FromIterator::from_iter(ids.into_iter().map(|i| self.remove_stream(i).unwrap())) } } @@ -229,7 +259,7 @@ pub struct DefaultSessionState where S: Stream { /// All streams that the session state is currently aware of. - streams: HashMap, + streams: HashMap>, /// The next available ID for outgoing streams. next_stream_id: StreamId, /// The parity bit for outgoing connections. Client-initiated connections must always be @@ -301,7 +331,7 @@ impl SessionState for DefaultSessionState fn insert_outgoing(&mut self, stream: Self::Stream) -> StreamId { let id = self.next_stream_id; - self.streams.insert(id, stream); + self.streams.insert(id, Entry::new(stream)); self.next_stream_id += 2; id } @@ -309,24 +339,23 @@ impl SessionState for DefaultSessionState fn insert_incoming(&mut self, stream_id: StreamId, stream: Self::Stream) -> Result<(), ()> { if self.validate_incoming_parity(stream_id) { // TODO(mlalic): Assert that the stream IDs are monotonically increasing! - self.streams.insert(stream_id, stream); + self.streams.insert(stream_id, Entry::new(stream)); Ok(()) } else { Err(()) } } - #[inline] - fn get_stream_ref(&self, stream_id: StreamId) -> Option<&Self::Stream> { + fn get_entry_ref(&self, stream_id: StreamId) -> Option<&Entry> { self.streams.get(&stream_id) } - #[inline] - fn get_stream_mut(&mut self, stream_id: StreamId) -> Option<&mut Self::Stream> { + + fn get_entry_mut(&mut self, stream_id: StreamId) -> Option<&mut Entry> { self.streams.get_mut(&stream_id) } #[inline] - fn remove_stream(&mut self, stream_id: StreamId) -> Option { + fn remove_stream(&mut self, stream_id: StreamId) -> Option> { self.streams.remove(&stream_id) } diff --git a/src/server/mod.rs b/src/server/mod.rs index c59323c3..d451a556 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -146,10 +146,9 @@ impl SimpleServer /// into the returned `Vec`. fn handle_requests(&mut self) -> HttpResult> { let handler = &mut self.handler; - let closed = self.conn - .state - .iter() - .filter(|&(_, ref s)| s.is_closed_remote()); + let closed = self.conn.state.iter() + .map(|(id, e)| (id, e.stream_mut())) + .filter(|&(_, ref s)| s.is_closed_remote()); let responses = closed.map(|(&stream_id, stream)| { let req = ServerRequest { stream_id: stream_id, From 64abe81c27c5899f86179d7e12d2e623ebdaa5f4 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 28 Mar 2016 01:39:10 +0000 Subject: [PATCH 09/24] Stream: Add `on_stream_error` method The method is used to signal errors that the local peer detects, in contrast to those that the remote signals by sending an RST_STREAM frame. The Session should send the RST_STREAM after the `on_stream_error` is called. --- src/http/session.rs | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/src/http/session.rs b/src/http/session.rs index 4e355ea6..2ed97b2d 100644 --- a/src/http/session.rs +++ b/src/http/session.rs @@ -425,9 +425,11 @@ pub enum StreamDataChunk { pub trait Stream { /// Handle a new data chunk that has arrived for the stream. fn new_data_chunk(&mut self, data: &[u8]); + /// Set headers for a stream. A stream is only allowed to have one set of /// headers. fn set_headers<'n, 'v>(&mut self, headers: Vec>); + /// Sets the stream state to the newly provided state. fn set_state(&mut self, state: StreamState); @@ -440,6 +442,14 @@ pub trait Stream { self.close(); } + /// Notifies the `Stream` that a stream error has been detected. This differs from + /// `on_rst_stream` in that the error was detected by the local peer, rather than the remote. + /// + /// The default implementation simply closes the stream. + fn on_stream_error(&mut self, _error_code: ErrorCode) { + self.close(); + } + /// Places the next data chunk that should be written onto the stream into the given buffer. /// /// # Returns @@ -463,6 +473,7 @@ pub trait Stream { fn close(&mut self) { self.set_state(StreamState::Closed); } + /// Updates the `Stream` status to indicate that it is closed locally. /// /// If the stream is closed on the remote end, then it is fully closed after this call. @@ -473,6 +484,7 @@ pub trait Stream { }; self.set_state(next); } + /// Updates the `Stream` status to indicate that it is closed on the remote peer's side. /// /// If the stream is also locally closed, then it is fully closed after this call. @@ -483,12 +495,14 @@ pub trait Stream { }; self.set_state(next); } + /// Returns whether the stream is closed. /// /// A stream is considered to be closed iff its state is set to `Closed`. fn is_closed(&self) -> bool { self.state() == StreamState::Closed } + /// Returns whether the stream is closed locally. fn is_closed_local(&self) -> bool { match self.state() { @@ -496,6 +510,7 @@ pub trait Stream { _ => false, } } + /// Returns whether the remote peer has closed the stream. This includes a fully closed stream. fn is_closed_remote(&self) -> bool { match self.state() { @@ -612,7 +627,7 @@ impl Stream for DefaultStream { #[cfg(test)] mod tests { use super::{Stream, DefaultSessionState, DefaultStream, StreamDataChunk, StreamDataError, - SessionState, Parity}; + SessionState, Parity, StreamState}; use super::Client as ClientMarker; use super::Server as ServerMarker; use http::{ErrorCode, Header}; @@ -826,6 +841,14 @@ mod tests { }); } + /// Tests that when the `DefaultStream` receives an error, it closes the stream. + #[test] + fn test_default_stream_on_error() { + let mut stream = DefaultStream::new(); + stream.on_stream_error(ErrorCode::FlowControlError); + assert_eq!(stream.state(), StreamState::Closed); + } + #[test] /// test_second_header_call will ensure that if headers are called twice in one stream (such as /// to set trailers) both results will be added to the stream's headers. From 3a911eceded7bad1e2a84593ad490147e05a8103 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 28 Mar 2016 02:47:56 +0000 Subject: [PATCH 10/24] Move http::client tests module to a standalone file The mod.rs file was getting quite large... --- src/http/client/mod.rs | 290 +-------------------------------------- src/http/client/tests.rs | 287 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 289 insertions(+), 288 deletions(-) create mode 100644 src/http/client/tests.rs diff --git a/src/http/client/mod.rs b/src/http/client/mod.rs index 65e210dc..0d8f1fb7 100644 --- a/src/http/client/mod.rs +++ b/src/http/client/mod.rs @@ -17,6 +17,8 @@ use http::priority::SimplePrioritizer; #[cfg(feature="tls")] pub mod tls; +#[cfg(test)] mod tests; + /// Writes the client preface to the given `io::Write` instance. /// /// According to the HTTP/2 spec, a client preface is first a specific sequence of octets, followed @@ -408,291 +410,3 @@ impl<'a, State, S> Session for ClientSession<'a, State, S> } } -#[cfg(test)] -mod tests { - use super::{ClientSession, write_preface, RequestStream}; - - use http::{Header, ErrorCode, HttpError}; - use http::tests::common::{TestStream, build_mock_client_conn, build_mock_http_conn, - MockReceiveFrame, MockSendFrame}; - use http::frame::{SettingsFrame, DataFrame, Frame, RawFrame}; - use http::connection::{HttpFrame, SendStatus}; - use http::session::{Session, SessionState, Stream, DefaultSessionState}; - use http::session::Client as ClientMarker; - - /// Tests that a client connection is correctly initialized, by reading the - /// server preface (i.e. a settings frame) as the first frame of the connection. - #[test] - fn test_init_client_conn() { - let frames = vec![HttpFrame::SettingsFrame(SettingsFrame::new())]; - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - let mut receiver = MockReceiveFrame::new(frames); - - conn.expect_settings(&mut receiver, &mut sender).unwrap(); - - // We have read the server's response (the settings frame only, since no panic - // ocurred) - assert_eq!(receiver.recv_list.len(), 0); - // We also sent an ACK already. - assert_eq!(sender.sent.len(), 1); - let frame = match HttpFrame::from_raw(&sender.sent[0]).unwrap() { - HttpFrame::SettingsFrame(frame) => frame, - _ => panic!("ACK not sent!"), - }; - assert!(frame.is_ack()); - } - - /// Tests that a client connection fails to initialize when the server does - /// not send a settings frame as its first frame (i.e. server preface). - #[test] - fn test_init_client_conn_no_settings() { - let frames = vec![HttpFrame::DataFrame(DataFrame::new(1))]; - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - let mut receiver = MockReceiveFrame::new(frames); - - // We get an error since the first frame sent by the server was not - // SETTINGS. - assert!(conn.expect_settings(&mut receiver, &mut sender).is_err()); - } - - /// A helper function that prepares a `TestStream` with an optional outgoing data stream. - fn prepare_stream(data: Option>) -> TestStream { - let mut stream = TestStream::new(); - match data { - None => stream.close_local(), - Some(d) => stream.set_outgoing(d), - }; - return stream; - } - - /// Tests that the `ClientConnection` correctly sends the next data, depending on the streams - /// known to it. - #[test] - fn test_client_conn_send_next_data() { - { - // No streams => nothing sent. - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Nothing); - } - { - // A locally closed stream (i.e. nothing to send) - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - conn.state.insert_outgoing(prepare_stream(None)); - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Nothing); - } - { - // A stream with some data - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Sent); - - // All of it got sent in the first go, so now we've got nothing? - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Nothing); - } - { - // Multiple streams with data - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); - conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); - conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); - for _ in 0..3 { - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Sent); - } - // All of it got sent in the first go, so now we've got nothing? - let res = conn.send_next_data(&mut sender).unwrap(); - assert_eq!(res, SendStatus::Nothing); - } - } - - /// Tests that the `ClientConnection::start_request` method correctly starts a new request. - #[test] - fn test_client_conn_start_request() { - { - // No body - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - - let stream = RequestStream { - headers: vec![ - Header::new(b":method", b"GET"), - ], - stream: prepare_stream(None), - }; - conn.start_request(stream, &mut sender).unwrap(); - - // The stream is in the connection state? - assert!(conn.state.get_stream_ref(1).is_some()); - // The headers got sent? - // (It'd be so much nicer to assert that the `send_headers` method got called) - assert_eq!(sender.sent.len(), 1); - match HttpFrame::from_raw(&sender.sent[0]).unwrap() { - HttpFrame::HeadersFrame(ref frame) => { - // The frame closed the stream? - assert!(frame.is_end_of_stream()); - } - _ => panic!("Expected a Headers frame"), - }; - } - { - // With a body - let mut conn = build_mock_client_conn(); - let mut sender = MockSendFrame::new(); - - let stream = RequestStream { - headers: vec![ - Header::new(b":method", b"POST"), - ], - stream: prepare_stream(Some(vec![1, 2, 3])), - }; - conn.start_request(stream, &mut sender).unwrap(); - - // The stream is in the connection state? - assert!(conn.state.get_stream_ref(1).is_some()); - // The headers got sent? - // (It'd be so much nicer to assert that the `send_headers` method got called) - assert_eq!(sender.sent.len(), 1); - match HttpFrame::from_raw(&sender.sent.remove(0)).unwrap() { - HttpFrame::HeadersFrame(ref frame) => { - // The stream is still open - assert!(!frame.is_end_of_stream()); - } - _ => panic!("Expected a Headers frame"), - }; - } - } - - /// Tests that a `ClientSession` notifies the correct stream when the - /// appropriate callback is invoked. - /// - /// A better unit test would give a mock Stream to the `ClientSession`, - /// instead of testing both the `ClientSession` and the `DefaultStream` - /// in the same time... - #[test] - fn test_client_session_notifies_stream() { - let mut state = DefaultSessionState::::new(); - state.insert_outgoing(TestStream::new()); - let mut conn = build_mock_http_conn(); - let mut sender = MockSendFrame::new(); - - { - // Registering some data to stream 1... - let mut session = ClientSession::new(&mut state, &mut sender); - session.new_data_chunk(1, &[1, 2, 3], &mut conn).unwrap(); - } - // ...works. - assert_eq!(state.get_stream_ref(1).unwrap().body, vec![1, 2, 3]); - { - // Some more... - let mut session = ClientSession::new(&mut state, &mut sender); - session.new_data_chunk(1, &[4], &mut conn).unwrap(); - } - // ...works. - assert_eq!(state.get_stream_ref(1).unwrap().body, vec![1, 2, 3, 4]); - // Now headers? - let headers = vec![ - Header::new(b":method", b"GET"), - ]; - { - let mut session = ClientSession::new(&mut state, &mut sender); - session.new_headers(1, headers.clone(), &mut conn).unwrap(); - } - assert_eq!(state.get_stream_ref(1).unwrap().headers.clone().unwrap(), - headers); - // Add another stream in the mix - state.insert_outgoing(TestStream::new()); - { - // and send it some data - let mut session = ClientSession::new(&mut state, &mut sender); - session.new_data_chunk(3, &[100], &mut conn).unwrap(); - } - assert_eq!(state.get_stream_ref(3).unwrap().body, vec![100]); - { - // Finally, the stream 1 ends... - let mut session = ClientSession::new(&mut state, &mut sender); - session.end_of_stream(1, &mut conn).unwrap(); - } - // ...and gets closed. - assert!(state.get_stream_ref(1).unwrap().is_closed()); - // but not the other one. - assert!(!state.get_stream_ref(3).unwrap().is_closed()); - // Sanity check: both streams still found in the session - assert_eq!(state.iter().collect::>().len(), 2); - // The closed stream is returned... - let closed = state.get_closed(); - assert_eq!(closed.len(), 1); - // ...and is also removed from the session! - assert_eq!(state.iter().collect::>().len(), 1); - } - - /// Tests that the `ClientSession` notifies the correct stream when it is reset by the peer. - #[test] - fn test_client_session_on_rst_stream() { - let mut state = DefaultSessionState::::new(); - state.insert_outgoing(TestStream::new()); - state.insert_outgoing(TestStream::new()); - let mut conn = build_mock_http_conn(); - let mut sender = MockSendFrame::new(); - { - let mut session = ClientSession::new(&mut state, &mut sender); - session.rst_stream(3, ErrorCode::Cancel, &mut conn).unwrap(); - } - assert!(state.get_stream_ref(3) - .map(|stream| { - stream.errors.len() == 1 && stream.errors[0] == ErrorCode::Cancel - }) - .unwrap()); - assert!(state.get_stream_ref(1).map(|stream| stream.errors.len() == 0).unwrap()); - } - - /// Tests that the `ClientSession` signals the correct error to client code when told to go - /// away by the peer. - #[test] - fn test_client_session_on_goaway() { - let mut state = DefaultSessionState::::new(); - let mut conn = build_mock_http_conn(); - let mut sender = MockSendFrame::new(); - let res = { - let mut session = ClientSession::new(&mut state, &mut sender); - session.on_goaway(0, ErrorCode::ProtocolError, None, &mut conn) - }; - if let Err(HttpError::PeerConnectionError(err)) = res { - assert_eq!(err.error_code(), ErrorCode::ProtocolError); - assert_eq!(err.debug_data(), None); - } else { - panic!("Expected a PeerConnectionError"); - } - } - - /// Tests that the `write_preface` function correctly writes a client preface to - /// a given `io::Write`. - #[test] - fn test_write_preface() { - // The buffer (`io::Write`) into which we will write the preface. - let mut written: Vec = Vec::new(); - - // Do it... - write_preface(&mut written).unwrap(); - - // The first bytes written to the underlying transport layer are the - // preface bytes. - let preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; - let frames_buf = &written[preface.len()..]; - // Immediately after that we sent a settings frame... - assert_eq!(preface, &written[..preface.len()]); - let raw = RawFrame::parse(frames_buf).unwrap(); - let frame: SettingsFrame = Frame::from_raw(&raw).unwrap(); - // ...which was not an ack, but our own settings. - assert!(!frame.is_ack()); - } -} diff --git a/src/http/client/tests.rs b/src/http/client/tests.rs new file mode 100644 index 00000000..d27564c4 --- /dev/null +++ b/src/http/client/tests.rs @@ -0,0 +1,287 @@ +//! Tests for the `http::client` module + +use super::{ClientSession, write_preface, RequestStream}; + +use http::{Header, ErrorCode, HttpError}; +use http::tests::common::{TestStream, build_mock_client_conn, build_mock_http_conn, + MockReceiveFrame, MockSendFrame}; +use http::frame::{SettingsFrame, DataFrame, Frame, RawFrame}; +use http::connection::{HttpFrame, SendStatus}; +use http::session::{Session, SessionState, Stream, DefaultSessionState}; +use http::session::Client as ClientMarker; + +/// Tests that a client connection is correctly initialized, by reading the +/// server preface (i.e. a settings frame) as the first frame of the connection. +#[test] +fn test_init_client_conn() { + let frames = vec![HttpFrame::SettingsFrame(SettingsFrame::new())]; + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + let mut receiver = MockReceiveFrame::new(frames); + + conn.expect_settings(&mut receiver, &mut sender).unwrap(); + + // We have read the server's response (the settings frame only, since no panic + // ocurred) + assert_eq!(receiver.recv_list.len(), 0); + // We also sent an ACK already. + assert_eq!(sender.sent.len(), 1); + let frame = match HttpFrame::from_raw(&sender.sent[0]).unwrap() { + HttpFrame::SettingsFrame(frame) => frame, + _ => panic!("ACK not sent!"), + }; + assert!(frame.is_ack()); +} + +/// Tests that a client connection fails to initialize when the server does +/// not send a settings frame as its first frame (i.e. server preface). +#[test] +fn test_init_client_conn_no_settings() { + let frames = vec![HttpFrame::DataFrame(DataFrame::new(1))]; + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + let mut receiver = MockReceiveFrame::new(frames); + + // We get an error since the first frame sent by the server was not + // SETTINGS. + assert!(conn.expect_settings(&mut receiver, &mut sender).is_err()); +} + +/// A helper function that prepares a `TestStream` with an optional outgoing data stream. +fn prepare_stream(data: Option>) -> TestStream { + let mut stream = TestStream::new(); + match data { + None => stream.close_local(), + Some(d) => stream.set_outgoing(d), + }; + return stream; +} + +/// Tests that the `ClientConnection` correctly sends the next data, depending on the streams +/// known to it. +#[test] +fn test_client_conn_send_next_data() { + { + // No streams => nothing sent. + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Nothing); + } + { + // A locally closed stream (i.e. nothing to send) + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + conn.state.insert_outgoing(prepare_stream(None)); + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Nothing); + } + { + // A stream with some data + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Sent); + + // All of it got sent in the first go, so now we've got nothing? + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Nothing); + } + { + // Multiple streams with data + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); + conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); + conn.state.insert_outgoing(prepare_stream(Some(vec![1, 2, 3]))); + for _ in 0..3 { + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Sent); + } + // All of it got sent in the first go, so now we've got nothing? + let res = conn.send_next_data(&mut sender).unwrap(); + assert_eq!(res, SendStatus::Nothing); + } +} + +/// Tests that the `ClientConnection::start_request` method correctly starts a new request. +#[test] +fn test_client_conn_start_request() { + { + // No body + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + + let stream = RequestStream { + headers: vec![ + Header::new(b":method", b"GET"), + ], + stream: prepare_stream(None), + }; + conn.start_request(stream, &mut sender).unwrap(); + + // The stream is in the connection state? + assert!(conn.state.get_stream_ref(1).is_some()); + // The headers got sent? + // (It'd be so much nicer to assert that the `send_headers` method got called) + assert_eq!(sender.sent.len(), 1); + match HttpFrame::from_raw(&sender.sent[0]).unwrap() { + HttpFrame::HeadersFrame(ref frame) => { + // The frame closed the stream? + assert!(frame.is_end_of_stream()); + } + _ => panic!("Expected a Headers frame"), + }; + } + { + // With a body + let mut conn = build_mock_client_conn(); + let mut sender = MockSendFrame::new(); + + let stream = RequestStream { + headers: vec![ + Header::new(b":method", b"POST"), + ], + stream: prepare_stream(Some(vec![1, 2, 3])), + }; + conn.start_request(stream, &mut sender).unwrap(); + + // The stream is in the connection state? + assert!(conn.state.get_stream_ref(1).is_some()); + // The headers got sent? + // (It'd be so much nicer to assert that the `send_headers` method got called) + assert_eq!(sender.sent.len(), 1); + match HttpFrame::from_raw(&sender.sent.remove(0)).unwrap() { + HttpFrame::HeadersFrame(ref frame) => { + // The stream is still open + assert!(!frame.is_end_of_stream()); + } + _ => panic!("Expected a Headers frame"), + }; + } +} + +/// Tests that a `ClientSession` notifies the correct stream when the +/// appropriate callback is invoked. +/// +/// A better unit test would give a mock Stream to the `ClientSession`, +/// instead of testing both the `ClientSession` and the `DefaultStream` +/// in the same time... +#[test] +fn test_client_session_notifies_stream() { + let mut state = DefaultSessionState::::new(); + state.insert_outgoing(TestStream::new()); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + + { + // Registering some data to stream 1... + let mut session = ClientSession::new(&mut state, &mut sender); + session.new_data_chunk(1, &[1, 2, 3], &mut conn).unwrap(); + } + // ...works. + assert_eq!(state.get_stream_ref(1).unwrap().body, vec![1, 2, 3]); + { + // Some more... + let mut session = ClientSession::new(&mut state, &mut sender); + session.new_data_chunk(1, &[4], &mut conn).unwrap(); + } + // ...works. + assert_eq!(state.get_stream_ref(1).unwrap().body, vec![1, 2, 3, 4]); + // Now headers? + let headers = vec![ + Header::new(b":method", b"GET"), + ]; + { + let mut session = ClientSession::new(&mut state, &mut sender); + session.new_headers(1, headers.clone(), &mut conn).unwrap(); + } + assert_eq!(state.get_stream_ref(1).unwrap().headers.clone().unwrap(), + headers); + // Add another stream in the mix + state.insert_outgoing(TestStream::new()); + { + // and send it some data + let mut session = ClientSession::new(&mut state, &mut sender); + session.new_data_chunk(3, &[100], &mut conn).unwrap(); + } + assert_eq!(state.get_stream_ref(3).unwrap().body, vec![100]); + { + // Finally, the stream 1 ends... + let mut session = ClientSession::new(&mut state, &mut sender); + session.end_of_stream(1, &mut conn).unwrap(); + } + // ...and gets closed. + assert!(state.get_stream_ref(1).unwrap().is_closed()); + // but not the other one. + assert!(!state.get_stream_ref(3).unwrap().is_closed()); + // Sanity check: both streams still found in the session + assert_eq!(state.iter().collect::>().len(), 2); + // The closed stream is returned... + let closed = state.get_closed(); + assert_eq!(closed.len(), 1); + // ...and is also removed from the session! + assert_eq!(state.iter().collect::>().len(), 1); +} + +/// Tests that the `ClientSession` notifies the correct stream when it is reset by the peer. +#[test] +fn test_client_session_on_rst_stream() { + let mut state = DefaultSessionState::::new(); + state.insert_outgoing(TestStream::new()); + state.insert_outgoing(TestStream::new()); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + { + let mut session = ClientSession::new(&mut state, &mut sender); + session.rst_stream(3, ErrorCode::Cancel, &mut conn).unwrap(); + } + assert!(state.get_stream_ref(3) + .map(|stream| { + stream.errors.len() == 1 && stream.errors[0] == ErrorCode::Cancel + }) + .unwrap()); + assert!(state.get_stream_ref(1).map(|stream| stream.errors.len() == 0).unwrap()); +} + +/// Tests that the `ClientSession` signals the correct error to client code when told to go +/// away by the peer. +#[test] +fn test_client_session_on_goaway() { + let mut state = DefaultSessionState::::new(); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + let res = { + let mut session = ClientSession::new(&mut state, &mut sender); + session.on_goaway(0, ErrorCode::ProtocolError, None, &mut conn) + }; + if let Err(HttpError::PeerConnectionError(err)) = res { + assert_eq!(err.error_code(), ErrorCode::ProtocolError); + assert_eq!(err.debug_data(), None); + } else { + panic!("Expected a PeerConnectionError"); + } +} + +/// Tests that the `write_preface` function correctly writes a client preface to +/// a given `io::Write`. +#[test] +fn test_write_preface() { + // The buffer (`io::Write`) into which we will write the preface. + let mut written: Vec = Vec::new(); + + // Do it... + write_preface(&mut written).unwrap(); + + // The first bytes written to the underlying transport layer are the + // preface bytes. + let preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; + let frames_buf = &written[preface.len()..]; + // Immediately after that we sent a settings frame... + assert_eq!(preface, &written[..preface.len()]); + let raw = RawFrame::parse(frames_buf).unwrap(); + let frame: SettingsFrame = Frame::from_raw(&raw).unwrap(); + // ...which was not an ack, but our own settings. + assert!(!frame.is_ack()); +} From 571ed4512d94bd0b36d9a6eb9e8158ae1420f8b8 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 28 Mar 2016 06:45:41 +0000 Subject: [PATCH 11/24] Add WindowUpdateStrategy trait and an impl that disables flow control The WindowUpdateStrategy trait is used to define the algorithm for updating the flow control window of both the connection, as well as individual stream. The object is provided the new size of the corresponding window and is expected to produce an action to be taken immediately: either increase the size of the window or do nothing. --- src/http/flow_control.rs | 130 +++++++++++++++++++++++++++++++++++++++ src/http/mod.rs | 1 + 2 files changed, 131 insertions(+) create mode 100644 src/http/flow_control.rs diff --git a/src/http/flow_control.rs b/src/http/flow_control.rs new file mode 100644 index 00000000..747370fb --- /dev/null +++ b/src/http/flow_control.rs @@ -0,0 +1,130 @@ +//! The module exposes the `WindowUpdateStrategy` trait, a trait used for defining how a flow +//! control window should be updated. +//! +//! A basic implementation of this trait which effectively disables flow control is also provided. + +use http::{StreamId, WindowSize, DEFAULT_MAX_WINDOW_SIZE}; + +/// Defines the possible actions that can be taken to update flow control. +#[derive(Debug, Copy, Clone, PartialEq)] +pub enum WindowUpdateAction { + /// Take no action -- do not change flow control + NoAction, + /// Increase the flow control window by the given amount + Increment(u32), +} + +/// A trait that should be implemented by types that are able to serve as algorithms for flow +/// control. +pub trait WindowUpdateStrategy { + /// Return the action that should be taken with respect to the connection-level flow control + /// window, depending on the given old and new window size. + fn on_connection_window(&mut self, + new: WindowSize) + -> WindowUpdateAction; + + /// Return the action that should be taken with respect to a stream-level flow control window, + /// depending on the given old and new window size for the stream. + fn on_stream_window(&mut self, + stream_id: StreamId, + new: WindowSize) + -> WindowUpdateAction; +} + +/// Provides an implementation of the `WindowUpdateStrategy` trait which effectively disables flow +/// control by updating the window sizes immediately after each decrease. +pub struct NoFlowControlStrategy { + max_connection_window_size: WindowSize, + max_stream_window_size: WindowSize, +} + +impl NoFlowControlStrategy { + /// Creates a new `NoFlowControlStrategy` which uses the default maximum window size for both + /// the connection, as well as the stream windows. + pub fn new() -> NoFlowControlStrategy { + NoFlowControlStrategy::with_max_window_sizes(DEFAULT_MAX_WINDOW_SIZE, + DEFAULT_MAX_WINDOW_SIZE) + } + + /// Creates a new `NoFlowControlStrategy` with the given maximum size of the connection and + /// stream windows. + pub fn with_max_window_sizes(max_connection: WindowSize, + max_stream: WindowSize) + -> NoFlowControlStrategy { + NoFlowControlStrategy { + max_connection_window_size: max_connection, + max_stream_window_size: max_stream, + } + } + + /// A private helper function that decides the action depending on the current window size and + /// the maximum window size. The action is always such that the current size gets increased to + /// the maximum size. + fn compute_update(current: WindowSize, max: WindowSize) -> WindowUpdateAction { + let current: i32 = current.into(); + let max: i32 = max.into(); + let delta = max - current; + if delta <= 0 { + WindowUpdateAction::NoAction + } else { + WindowUpdateAction::Increment(delta as u32) + } + } +} + +impl WindowUpdateStrategy for NoFlowControlStrategy { + #[inline] + fn on_connection_window(&mut self, + new: WindowSize) + -> WindowUpdateAction { + NoFlowControlStrategy::compute_update(new, self.max_connection_window_size) + } + + #[inline] + fn on_stream_window(&mut self, + _stream_id: StreamId, + new: WindowSize) + -> WindowUpdateAction { + NoFlowControlStrategy::compute_update(new, self.max_stream_window_size) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use http::WindowSize; + + #[test] + fn test_window_update_no_flow_ctrl() { + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_connection_window(WindowSize(0)); + assert_eq!(action, WindowUpdateAction::Increment(0xffff)); + } + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_connection_window(WindowSize(1)); + assert_eq!(action, WindowUpdateAction::Increment(0xffff - 1)); + } + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_connection_window(WindowSize(0xffff + 1)); + assert_eq!(action, WindowUpdateAction::NoAction); + } + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_stream_window(1, WindowSize(0)); + assert_eq!(action, WindowUpdateAction::Increment(0xffff)); + } + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_stream_window(1, WindowSize(1)); + assert_eq!(action, WindowUpdateAction::Increment(0xffff - 1)); + } + { + let mut strat = NoFlowControlStrategy::new(); + let action = strat.on_stream_window(1, WindowSize(0xffff + 1)); + assert_eq!(action, WindowUpdateAction::NoAction); + } + } +} diff --git a/src/http/mod.rs b/src/http/mod.rs index e55995f7..7919c808 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -13,6 +13,7 @@ pub mod transport; pub mod connection; pub mod session; pub mod priority; +pub mod flow_control; pub mod client; pub mod server; From 434a1d55277be9f6fb49276076400785e8b53da7 Mon Sep 17 00:00:00 2001 From: Marko Lalic Date: Mon, 28 Mar 2016 07:04:26 +0000 Subject: [PATCH 12/24] Flow control: Implement client-side inbound flow control --- src/http/client/mod.rs | 90 +++++++++++++++++++++++++ src/http/client/tests.rs | 141 ++++++++++++++++++++++++++++++++++++++- src/http/connection.rs | 11 +++ src/http/flow_control.rs | 5 +- src/http/session.rs | 14 +++- 5 files changed, 255 insertions(+), 6 deletions(-) diff --git a/src/http/client/mod.rs b/src/http/client/mod.rs index 0d8f1fb7..8d759b6d 100644 --- a/src/http/client/mod.rs +++ b/src/http/client/mod.rs @@ -13,6 +13,7 @@ use http::connection::{SendFrame, ReceiveFrame, SendStatus, HttpConnection, EndS use http::session::{Session, Stream, DefaultStream, DefaultSessionState, SessionState}; use http::session::Client as ClientMarker; use http::priority::SimplePrioritizer; +use http::flow_control::{WindowUpdateStrategy, NoFlowControlStrategy, WindowUpdateAction}; #[cfg(feature="tls")] pub mod tls; @@ -304,6 +305,7 @@ pub struct ClientSession<'a, State, S> { state: &'a mut State, sender: &'a mut S, + window_update_strategy: Option<&'a mut WindowUpdateStrategy>, } impl<'a, State, S> ClientSession<'a, State, S> @@ -316,6 +318,20 @@ impl<'a, State, S> ClientSession<'a, State, S> ClientSession { state: state, sender: sender, + window_update_strategy: None, + } + } + + /// Creates a new `ClientSession` associated to the given state, which uses the given + /// `WindowUpdateStrategy` to decide how the flow control windows should be updated. + pub fn with_window_update_strategy(state: &'a mut State, + sender: &'a mut S, + window_update_strategy: &'a mut WindowUpdateStrategy) + -> ClientSession<'a, State, S> { + ClientSession { + state: state, + sender: sender, + window_update_strategy: Some(window_update_strategy), } } } @@ -342,6 +358,7 @@ impl<'a, State, S> Session for ClientSession<'a, State, S> }; // Now let the stream handle the data chunk stream.new_data_chunk(data); + Ok(()) } @@ -408,5 +425,78 @@ impl<'a, State, S> Session for ClientSession<'a, State, S> debug!("Received a PING ack"); Ok(()) } + + fn on_connection_in_window_decrease(&mut self, conn: &mut HttpConnection) -> HttpResult<()> { + // The default reaction to the inbound window decreasing is to ask the window update + // strategy what should be done; if the window should be increased, emit the + // appropriate window update frame. + + let new = conn.in_window_size(); + let conn_update = match self.window_update_strategy.as_mut() { + Some(strategy) => { + strategy.on_connection_window(new) + } + None => { + NoFlowControlStrategy::new().on_connection_window(new) + } + }; + if let WindowUpdateAction::Increment(delta) = conn_update { + try!(conn.increase_connection_window_size(delta)); + try!(conn.sender(self.sender).send_connection_window_update(delta)); + } + + Ok(()) + } + + fn on_stream_in_window_decrease(&mut self, + stream_id: StreamId, + size: u32, + conn: &mut HttpConnection) + -> HttpResult<()> { + // TODO: Get rid of the assert. + debug_assert!(size <= 0x7fffffff); + let size: i32 = size as i32; + + if let Some(e) = self.state.get_entry_mut(stream_id) { + if e.inbound_window().can_accept(size) { + // First do the actual window update... + let old = *e.inbound_window(); + e.inbound_window_mut().try_decrease(size) + .ok() + .expect("Already checked that no overflow can happen"); + let new = *e.inbound_window(); + trace!("Stream window update: stream_id={:?}, old={:?}, new={:?}", + stream_id, + old, + new); + + // Now, check if the window update strategy mandates an increase in the window + // size. + let stream_update = match self.window_update_strategy.as_mut() { + Some(mut strategy) => { + strategy.on_stream_window(stream_id, new) + } + None => { + NoFlowControlStrategy::new().on_stream_window(stream_id, new) + } + }; + if let WindowUpdateAction::Increment(delta) = stream_update { + if let Err(_) = e.inbound_window_mut().try_increase(delta) { + // TODO: Should we perhaps propagate this error upward? + warn!("Misbehaving WindowUpdateStrategy would overflow the window size"); + } else { + try!(conn.sender(self.sender).send_stream_window_update(stream_id, delta)); + } + } + } else { + // This would violate the flow control for the stream. + let err = ErrorCode::FlowControlError; + e.stream_mut().on_stream_error(err); + try!(conn.sender(self.sender).rst_stream(stream_id, err)); + } + } + + Ok(()) + } } diff --git a/src/http/client/tests.rs b/src/http/client/tests.rs index d27564c4..ba85dcde 100644 --- a/src/http/client/tests.rs +++ b/src/http/client/tests.rs @@ -2,13 +2,15 @@ use super::{ClientSession, write_preface, RequestStream}; -use http::{Header, ErrorCode, HttpError}; +use http::{Header, ErrorCode, HttpError, WindowSize, StreamId}; use http::tests::common::{TestStream, build_mock_client_conn, build_mock_http_conn, MockReceiveFrame, MockSendFrame}; use http::frame::{SettingsFrame, DataFrame, Frame, RawFrame}; use http::connection::{HttpFrame, SendStatus}; -use http::session::{Session, SessionState, Stream, DefaultSessionState}; +use http::connection::{set_connection_windows}; +use http::session::{Session, SessionState, Stream, DefaultSessionState, StreamState}; use http::session::Client as ClientMarker; +use http::flow_control::{WindowUpdateStrategy, WindowUpdateAction}; /// Tests that a client connection is correctly initialized, by reading the /// server preface (i.e. a settings frame) as the first frame of the connection. @@ -285,3 +287,138 @@ fn test_write_preface() { // ...which was not an ack, but our own settings. assert!(!frame.is_ack()); } + +/// An implementation of the `WindowUpdateStrategy` trait that never takes any action. +struct NoActionStrategy; + +impl WindowUpdateStrategy for NoActionStrategy { + fn on_connection_window(&mut self, _: WindowSize) -> WindowUpdateAction { + WindowUpdateAction::NoAction + } + + fn on_stream_window(&mut self, + _: StreamId, + _: WindowSize) + -> WindowUpdateAction { + WindowUpdateAction::NoAction + } +} + +/// Tests that the `ClientSession` correctly updates the state +#[test] +fn test_session_window_decrease_state_update_valid() { + let mut state = DefaultSessionState::::new(); + // Prepare a stream in the state of the session, which will receive an update + let stream_id = state.insert_outgoing(TestStream::new()); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + + let decrease = 50; + let res = { + let mut no_action = NoActionStrategy; + let mut session = ClientSession::with_window_update_strategy(&mut state, + &mut sender, + &mut no_action); + session.on_stream_in_window_decrease(stream_id, decrease, &mut conn) + }; + + assert!(res.is_ok()); + let entry = state.get_entry_ref(stream_id).unwrap(); + assert_eq!(entry.inbound_window().size(), + 0xffff_i32 - decrease as i32); + assert_eq!(entry.stream_ref().state(), StreamState::Open); +} + +#[test] +fn test_session_window_decrease_state_update_window_too_small() { + let mut state = DefaultSessionState::::new(); + // Prepare a stream in the state of the session, which will receive an update + let stream_id = state.insert_outgoing(TestStream::new()); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + + let decrease = 0xffff + 1; + let res = { + let mut no_action = NoActionStrategy; + let mut session = ClientSession::with_window_update_strategy(&mut state, + &mut sender, + &mut no_action); + session.on_stream_in_window_decrease(stream_id, decrease, &mut conn) + }; + + // The window wasn't large enough to accomodate the update => RST_STREAM + assert!(res.is_ok()); + match HttpFrame::from_raw(&sender.sent.remove(0)).expect("RST_STREAM frame") { + HttpFrame::RstStreamFrame(ref frame) => { + assert_eq!(frame.get_stream_id(), stream_id); + } + _ => panic!("Expected an RST_STREAM frame"), + }; + // The stream got closed too. + assert_eq!(state.get_stream_ref(stream_id).unwrap().state(), StreamState::Closed); +} + +#[test] +fn test_session_flow_control_disabled_by_default_stream() { + let mut state = DefaultSessionState::::new(); + // Prepare a stream in the state of the session, which will receive an update + let stream_id = state.insert_outgoing(TestStream::new()); + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + + let decrease = 50; + let res = { + let mut session = ClientSession::new(&mut state, + &mut sender); + session.on_stream_in_window_decrease(stream_id, decrease, &mut conn) + }; + + assert!(res.is_ok()); + // The window got updated immediately. + let entry = state.get_entry_ref(stream_id).unwrap(); + assert_eq!(entry.inbound_window().size(), + 0xffff_i32); + assert_eq!(entry.stream_ref().state(), StreamState::Open); + // The window update frame is there + match HttpFrame::from_raw(&sender.sent.remove(0)).expect("WINDOW_UPDATE frame") { + HttpFrame::WindowUpdateFrame(ref frame) => { + assert_eq!(frame.get_stream_id(), stream_id); + assert_eq!(frame.increment(), decrease); + } + _ => panic!("Expected a WINDOW_UPDATE frame"), + }; +} + +#[test] +fn test_session_flow_control_disabled_by_default_connection() { + let mut state = DefaultSessionState::::new(); + // Prepare a stream in the state of the session, which will receive an update + let mut conn = build_mock_http_conn(); + let mut sender = MockSendFrame::new(); + + let decrease = 50; + let res = { + let mut session = ClientSession::new(&mut state, + &mut sender); + let (new_in_window, old_out_window) = { + let mut window = conn.in_window_size(); + window.try_decrease(decrease).unwrap(); + (window, conn.out_window_size()) + }; + set_connection_windows(&mut conn, new_in_window, old_out_window); + // Sanity check: the in window is less than the default + assert_eq!(conn.in_window_size(), 0xffff_i32 - decrease); + session.on_connection_in_window_decrease(&mut conn) + }; + + assert!(res.is_ok()); + // The window size got increased automatically by the delta... + assert_eq!(conn.in_window_size(), 0xffff_i32); + match HttpFrame::from_raw(&sender.sent.remove(0)).expect("WINDOW_UPDATE frame") { + HttpFrame::WindowUpdateFrame(ref frame) => { + assert_eq!(frame.get_stream_id(), 0); + assert_eq!(frame.increment(), decrease as u32); + } + _ => panic!("Expected a WINDOW_UPDATE frame"), + }; +} diff --git a/src/http/connection.rs b/src/http/connection.rs index 4636366c..c4dea1ad 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -622,6 +622,17 @@ impl HttpConnection { } } +/// A helper method for tests that allows the window sizes of the given connection to be modified. +/// Since this touches the internal state that isn't intended to be modified by clients directly, +/// it is intended only as a helper for tests. +#[cfg(test)] +pub fn set_connection_windows(conn: &mut HttpConnection, + in_window: WindowSize, + out_window: WindowSize) { + conn.in_window_size = in_window; + conn.out_window_size = out_window; +} + #[cfg(test)] mod tests { use std::borrow::Cow; diff --git a/src/http/flow_control.rs b/src/http/flow_control.rs index 747370fb..e554e412 100644 --- a/src/http/flow_control.rs +++ b/src/http/flow_control.rs @@ -18,13 +18,12 @@ pub enum WindowUpdateAction { /// control. pub trait WindowUpdateStrategy { /// Return the action that should be taken with respect to the connection-level flow control - /// window, depending on the given old and new window size. + /// window. fn on_connection_window(&mut self, new: WindowSize) -> WindowUpdateAction; - /// Return the action that should be taken with respect to a stream-level flow control window, - /// depending on the given old and new window size for the stream. + /// Return the action that should be taken with respect to a stream-level flow control window. fn on_stream_window(&mut self, stream_id: StreamId, new: WindowSize) diff --git a/src/http/session.rs b/src/http/session.rs index 2ed97b2d..50e257f0 100644 --- a/src/http/session.rs +++ b/src/http/session.rs @@ -9,7 +9,8 @@ use std::error::Error; use std::io::Read; use std::io::Cursor; use std::iter::FromIterator; -use http::{StreamId, OwnedHeader, Header, HttpResult, ErrorCode, HttpError, ConnectionError}; +use http::{StreamId, OwnedHeader, Header, HttpResult, ErrorCode, HttpError, ConnectionError, + DEFAULT_MAX_WINDOW_SIZE, WindowSize}; use http::frame::{HttpSetting, PingFrame}; use http::connection::HttpConnection; @@ -142,6 +143,10 @@ impl<'a, S> Iterator for StreamIter<'a, S> /// information for a single HTTP/2 stream. pub struct Entry where S: Stream { stream: S, + /// Tracks the size of the outbound flow control window + out_window: WindowSize, + /// Tracks the size of the inbound flow control window + in_window: WindowSize, } impl Entry where S: Stream { @@ -149,6 +154,9 @@ impl Entry where S: Stream { pub fn new(stream: S) -> Entry { Entry { stream: stream, + // TODO: Use the current initial window sizes indicated by the connection settings! + out_window: DEFAULT_MAX_WINDOW_SIZE, + in_window: DEFAULT_MAX_WINDOW_SIZE, } } /// Consumes the `Entry`, returning the underlying `Stream` instance @@ -157,6 +165,10 @@ impl Entry where S: Stream { pub fn stream_ref(&self) -> &S { &self.stream } /// Returns a mutable reference to the `Stream` pub fn stream_mut(&mut self) -> &mut S { &mut self.stream } + + pub fn outbound_window(&self) -> &WindowSize { &self.out_window } + pub fn inbound_window(&self) -> &WindowSize { &self.in_window } + pub fn inbound_window_mut(&mut self) -> &mut WindowSize { &mut self.in_window } } /// A trait defining a set of methods for accessing and influencing an HTTP/2 session's state. From 8f550e5576e55a7d639ef3252bf9f28a12fb0712 Mon Sep 17 00:00:00 2001 From: Rohit Joshi Date: Mon, 10 Apr 2017 22:56:02 -0400 Subject: [PATCH 13/24] Update tls.rs --- src/http/client/tls.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 67a3be55..8ef94812 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -205,7 +205,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) - context.set_verify(SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); + context.set_verify(SSL_VERIFY_PEER , None); try!(context.set_CA_file(ca_file_path)); // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); From 1f3122536394ab66cfa69678a0306c47a113de1f Mon Sep 17 00:00:00 2001 From: Rohit Joshi Date: Mon, 10 Apr 2017 23:10:12 -0400 Subject: [PATCH 14/24] Update tls.rs --- src/http/client/tls.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 8ef94812..e62e629c 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -205,7 +205,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) - context.set_verify(SSL_VERIFY_PEER , None); + // context.set_verify(SSL_VERIFY_PEER , None); try!(context.set_CA_file(ca_file_path)); // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); From 43faa33627b6499eb05318c9b8a8644e64cc3033 Mon Sep 17 00:00:00 2001 From: ytr289 Date: Mon, 10 Apr 2017 23:25:45 -0400 Subject: [PATCH 15/24] added support for custom port + disabled verify peer --- src/http/client/tls.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index e62e629c..555adceb 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -16,7 +16,7 @@ //! //! // Connect to an HTTP/2 aware server //! let path = "/path/to/certs.pem"; -//! let connector = TlsConnector::new("http2bin.org", &path); +//! let connector = TlsConnector::new("http2bin.org", 443, &path); //! let mut client = SimpleClient::with_connector(connector).unwrap(); //! let response = client.get(b"/get", &[]).unwrap(); //! assert_eq!(response.stream_id, 1); @@ -66,7 +66,7 @@ use openssl::ssl::SslMethod; /// /// // Connect to an HTTP/2 aware server /// let path = "/path/to/certs.pem"; -/// let connector = TlsConnector::new("http2bin.org", &path); +/// let connector = TlsConnector::new("http2bin.org", 443, &path); /// let mut client = SimpleClient::with_connector(connector).unwrap(); /// let response = client.get(b"/get", &[]).unwrap(); /// assert_eq!(response.stream_id, 1); @@ -85,6 +85,7 @@ use openssl::ssl::SslMethod; /// ``` pub struct TlsConnector<'a, 'ctx> { pub host: &'a str, + pub port: u16, context: Http2TlsContext<'ctx>, } @@ -182,18 +183,20 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Creates a new `TlsConnector` that will create a new `SslContext` before /// trying to establish the TLS connection. The path to the CA file that the /// context will use needs to be provided. - pub fn new>(host: &'a str, ca_file_path: &'ctx P) -> TlsConnector<'a, 'ctx> { + pub fn new>(host: &'a str, port:u16, ca_file_path: &'ctx P) -> TlsConnector<'a, 'ctx> { TlsConnector { host: host, + port: port, context: Http2TlsContext::CertPath(ca_file_path.as_ref()), } } /// Creates a new `TlsConnector` that will use the provided context to /// create the `SslStream` that will back the HTTP/2 connection. - pub fn with_context(host: &'a str, context: &'ctx SslContext) -> TlsConnector<'a, 'ctx> { + pub fn with_context(host: &'a str, port:u16, context: &'ctx SslContext) -> TlsConnector<'a, 'ctx> { TlsConnector { host: host, + port: port, context: Http2TlsContext::Wrapped(context), } } @@ -222,7 +225,7 @@ impl<'a, 'ctx> HttpConnect for TlsConnector<'a, 'ctx> { fn connect(self) -> Result>, TlsConnectError> { // First, create a TCP connection to port 443 - let raw_tcp = try!(TcpStream::connect(&(self.host, 443))); + let raw_tcp = try!(TcpStream::connect(&(self.host, self.port))); // Now build the SSL instance, depending on which SSL context should be // used... let ssl = match self.context { From 9fcb59a18d196f2326f50a7e1edca5e78ffd6d33 Mon Sep 17 00:00:00 2001 From: ytr289 Date: Tue, 11 Apr 2017 00:24:42 -0400 Subject: [PATCH 16/24] enable optional verify peer --- src/http/client/tls.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 555adceb..f2c0a840 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -208,7 +208,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) - // context.set_verify(SSL_VERIFY_PEER , None); + context.set_verify(SSL_VERIFY_PEER , None); try!(context.set_CA_file(ca_file_path)); // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); From 2dcd4df14bbe59cfb084f43246d7a15f4aa3d9f8 Mon Sep 17 00:00:00 2001 From: ytr289 Date: Tue, 11 Apr 2017 00:35:45 -0400 Subject: [PATCH 17/24] enable optional verify peer --- src/http/client/tls.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index f2c0a840..1816c547 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -46,7 +46,7 @@ use http::{HttpScheme, ALPN_PROTOCOLS}; use super::{ClientStream, write_preface, HttpConnect, HttpConnectError}; use openssl::ssl::{Ssl, SslStream, SslContext}; -use openssl::ssl::{SSL_VERIFY_PEER, SSL_VERIFY_FAIL_IF_NO_PEER_CERT}; +use openssl::ssl::{SSL_VERIFY_PEER, SSL_VERIFY_NONE, SSL_VERIFY_FAIL_IF_NO_PEER_CERT}; use openssl::ssl::SSL_OP_NO_COMPRESSION; use openssl::ssl::error::SslError; use openssl::ssl::SslMethod; @@ -203,12 +203,16 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a default `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_default_context(ca_file_path: &Path) -> Result { + pub fn build_default_context(ca_file_path: &Path, verify_peer : bool) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) - context.set_verify(SSL_VERIFY_PEER , None); + if verify_peer == true { + context.set_verify(SSL_VERIFY_PEER |SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); + }else { + context.set_verify(SSL_VERIFY_NONE, None); + } try!(context.set_CA_file(ca_file_path)); // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); @@ -230,7 +234,7 @@ impl<'a, 'ctx> HttpConnect for TlsConnector<'a, 'ctx> { // used... let ssl = match self.context { Http2TlsContext::CertPath(path) => { - let ctx = try!(TlsConnector::build_default_context(&path)); + let ctx = try!(TlsConnector::build_default_context(&path, true)); try!(Ssl::new(&ctx)) } Http2TlsContext::Wrapped(ctx) => try!(Ssl::new(ctx)), From 2d3f8253b0681abc150b6d01da47fc27e37b2c2b Mon Sep 17 00:00:00 2001 From: Rohit Joshi Date: Tue, 11 Apr 2017 22:01:29 -0400 Subject: [PATCH 18/24] Update tls.rs --- src/http/client/tls.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 1816c547..831af46c 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -203,7 +203,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a default `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_default_context(ca_file_path: &Path, verify_peer : bool) -> Result { + pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: &Option, private_key: &Option) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); @@ -214,6 +214,12 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { context.set_verify(SSL_VERIFY_NONE, None); } try!(context.set_CA_file(ca_file_path)); + if cert_file.is_some() { + try!(context.set_certificate_file(cert_file.unwrap())); + } + if private_key.is_some() { + try!(context.set_private_key_file(private_key.unwrap())); + } // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); // The HTTP/2 protocol identifiers are constant at the library level... From e4a156845f627cf9e0338aae566f11af9f1582e6 Mon Sep 17 00:00:00 2001 From: Rohit Joshi Date: Tue, 11 Apr 2017 22:07:28 -0400 Subject: [PATCH 19/24] Update tls.rs --- src/http/client/tls.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 831af46c..0ad6c4bb 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -203,7 +203,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a default `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: &Option, private_key: &Option) -> Result { + pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: Option, private_key: Option) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); From cca37fb14b92976869718651be1a18d723315f0a Mon Sep 17 00:00:00 2001 From: Rohit Joshi Date: Tue, 11 Apr 2017 22:09:41 -0400 Subject: [PATCH 20/24] Update tls.rs --- src/http/client/tls.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 0ad6c4bb..d85be86b 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -240,7 +240,7 @@ impl<'a, 'ctx> HttpConnect for TlsConnector<'a, 'ctx> { // used... let ssl = match self.context { Http2TlsContext::CertPath(path) => { - let ctx = try!(TlsConnector::build_default_context(&path, true)); + let ctx = try!(TlsConnector::build_default_context(&path, true, None, None)); try!(Ssl::new(&ctx)) } Http2TlsContext::Wrapped(ctx) => try!(Ssl::new(ctx)), From c1b0f9af6e228c89fc38b1bb167cd379a83f3d6e Mon Sep 17 00:00:00 2001 From: ytr289 Date: Tue, 11 Apr 2017 22:19:39 -0400 Subject: [PATCH 21/24] adding support for cert/private key --- src/http/client/tls.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index d85be86b..db761957 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -36,7 +36,7 @@ use std::convert::AsRef; use std::net::TcpStream; -use std::path::Path; +use std::path::{Path,PathBuf}; use std::error; use std::fmt; use std::str; @@ -50,6 +50,7 @@ use openssl::ssl::{SSL_VERIFY_PEER, SSL_VERIFY_NONE, SSL_VERIFY_FAIL_IF_NO_PEER_ use openssl::ssl::SSL_OP_NO_COMPRESSION; use openssl::ssl::error::SslError; use openssl::ssl::SslMethod; +use openssl::x509::X509FileType; /// A struct implementing the functionality of establishing a TLS-backed TCP stream /// that can be used by an HTTP/2 connection. Takes care to set all the TLS options @@ -203,7 +204,7 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a default `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: Option, private_key: Option) -> Result { + pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: Option, private_key: Option) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); @@ -215,10 +216,10 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { } try!(context.set_CA_file(ca_file_path)); if cert_file.is_some() { - try!(context.set_certificate_file(cert_file.unwrap())); + try!(context.set_certificate_file(cert_file.unwrap(),X509FileType::PEM)); } if private_key.is_some() { - try!(context.set_private_key_file(private_key.unwrap())); + try!(context.set_private_key_file(private_key.unwrap(),X509FileType::PEM)); } // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); From bb5eaa247939863646444e8e4f5f8236d183196c Mon Sep 17 00:00:00 2001 From: ytr289 Date: Wed, 12 Apr 2017 10:09:36 -0400 Subject: [PATCH 22/24] PR 37 review comment --- src/http/client/tls.rs | 38 +++++++++++++++++++++++++++++++------- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index db761957..8125ea59 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -204,22 +204,46 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a default `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_default_context(ca_file_path: &Path, verify_peer : bool, cert_file: Option, private_key: Option) -> Result { + pub fn build_default_context(ca_file_path: &Path) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); + // This makes the certificate required (only VERIFY_PEER would mean optional) + context.set_verify(SSL_VERIFY_PEER |SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); + + try!(context.set_CA_file(ca_file_path)); + + // Compression is not allowed by the spec + context.set_options(SSL_OP_NO_COMPRESSION); + // The HTTP/2 protocol identifiers are constant at the library level... + context.set_npn_protocols(ALPN_PROTOCOLS); + + Ok(context) + } + + /// Builds up a `SslContext` instance wth TLS settings that the + /// HTTP/2 spec mandates. The path to the CA file needs to be provided. + pub fn build_context(ssl_method: Option, ca_file_path: Option, verify_peer : bool, cert_file_path: Option, private_key_file_path: Option) -> Result { + // HTTP/2 connections need to be on top of TLSv1.2 or newer. + let mut context = try!(SslContext::new(ssl_method.unwrap_or(SslMethod::Tlsv1_2))); + // This makes the certificate required (only VERIFY_PEER would mean optional) if verify_peer == true { context.set_verify(SSL_VERIFY_PEER |SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); }else { context.set_verify(SSL_VERIFY_NONE, None); } - try!(context.set_CA_file(ca_file_path)); - if cert_file.is_some() { - try!(context.set_certificate_file(cert_file.unwrap(),X509FileType::PEM)); + // set the CA file + if ca_file_path.is_some() { + try!(context.set_CA_file(ca_file_path.unwrap())); + } + // set the client cert + if cert_file_path.is_some() { + try!(context.set_certificate_file(cert_file_path.unwrap(),X509FileType::PEM)); } - if private_key.is_some() { - try!(context.set_private_key_file(private_key.unwrap(),X509FileType::PEM)); + //set the private key + if private_key_file_path.is_some() { + try!(context.set_private_key_file(private_key_file_path.unwrap(),X509FileType::PEM)); } // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); @@ -241,7 +265,7 @@ impl<'a, 'ctx> HttpConnect for TlsConnector<'a, 'ctx> { // used... let ssl = match self.context { Http2TlsContext::CertPath(path) => { - let ctx = try!(TlsConnector::build_default_context(&path, true, None, None)); + let ctx = try!(TlsConnector::build_default_context(&path)); try!(Ssl::new(&ctx)) } Http2TlsContext::Wrapped(ctx) => try!(Ssl::new(ctx)), From fd84a82c36c81817f1b3e8686eba26800c122481 Mon Sep 17 00:00:00 2001 From: ytr289 Date: Wed, 12 Apr 2017 10:18:48 -0400 Subject: [PATCH 23/24] reverting to origin for build_default_context --- src/http/client/tls.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 8125ea59..434e6ebe 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -209,10 +209,8 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) - context.set_verify(SSL_VERIFY_PEER |SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); - + context.set_verify(SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, None); try!(context.set_CA_file(ca_file_path)); - // Compression is not allowed by the spec context.set_options(SSL_OP_NO_COMPRESSION); // The HTTP/2 protocol identifiers are constant at the library level... From 0b559466bc3d042bdca4d52c24e7ce2968147d28 Mon Sep 17 00:00:00 2001 From: ytr289 Date: Wed, 12 Apr 2017 10:23:59 -0400 Subject: [PATCH 24/24] removing dependency on openssl for client. defaulted method to tlsv1.2 --- src/http/client/tls.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/http/client/tls.rs b/src/http/client/tls.rs index 434e6ebe..e819b410 100644 --- a/src/http/client/tls.rs +++ b/src/http/client/tls.rs @@ -221,9 +221,9 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { /// Builds up a `SslContext` instance wth TLS settings that the /// HTTP/2 spec mandates. The path to the CA file needs to be provided. - pub fn build_context(ssl_method: Option, ca_file_path: Option, verify_peer : bool, cert_file_path: Option, private_key_file_path: Option) -> Result { + pub fn build_context(ca_file_path: Option, verify_peer : bool, cert_file_path: Option, private_key_file_path: Option) -> Result { // HTTP/2 connections need to be on top of TLSv1.2 or newer. - let mut context = try!(SslContext::new(ssl_method.unwrap_or(SslMethod::Tlsv1_2))); + let mut context = try!(SslContext::new(SslMethod::Tlsv1_2)); // This makes the certificate required (only VERIFY_PEER would mean optional) if verify_peer == true {