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/client/mod.rs b/src/http/client/mod.rs index 65e210dc..8d759b6d 100644 --- a/src/http/client/mod.rs +++ b/src/http/client/mod.rs @@ -13,10 +13,13 @@ 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; +#[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 @@ -302,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> @@ -314,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), } } } @@ -340,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(()) } @@ -406,293 +425,78 @@ impl<'a, State, S> Session for ClientSession<'a, State, S> debug!("Received a PING ack"); Ok(()) } -} -#[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()); - } + 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. - /// 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); + let new = conn.in_window_size(); + let conn_update = match self.window_update_strategy.as_mut() { + Some(strategy) => { + strategy.on_connection_window(new) } - // 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(); + 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)); } - // ...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()); + Ok(()) } - /// 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"); + 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)); + } } - } - /// 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()); + Ok(()) } } + diff --git a/src/http/client/tests.rs b/src/http/client/tests.rs new file mode 100644 index 00000000..ba85dcde --- /dev/null +++ b/src/http/client/tests.rs @@ -0,0 +1,424 @@ +//! Tests for the `http::client` module + +use super::{ClientSession, write_preface, RequestStream}; + +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::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. +#[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()); +} + +/// 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/client/tls.rs b/src/http/client/tls.rs index 67a3be55..e819b410 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); @@ -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; @@ -46,10 +46,11 @@ 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; +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 @@ -66,7 +67,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 +86,7 @@ use openssl::ssl::SslMethod; /// ``` pub struct TlsConnector<'a, 'ctx> { pub host: &'a str, + pub port: u16, context: Http2TlsContext<'ctx>, } @@ -182,18 +184,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), } } @@ -214,6 +218,38 @@ impl<'a, 'ctx> TlsConnector<'a, 'ctx> { 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(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(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); + } + // 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)); + } + //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); + // The HTTP/2 protocol identifiers are constant at the library level... + context.set_npn_protocols(ALPN_PROTOCOLS); + + Ok(context) + } } impl<'a, 'ctx> HttpConnect for TlsConnector<'a, 'ctx> { @@ -222,7 +258,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 { diff --git a/src/http/connection.rs b/src/http/connection.rs index 18fcfdcf..c4dea1ad 100644 --- a/src/http/connection.rs +++ b/src/http/connection.rs @@ -274,11 +274,34 @@ 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) } + /// 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 @@ -352,13 +375,23 @@ 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 + } + + /// 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 @@ -452,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); @@ -471,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()); + 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()); @@ -532,7 +571,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)); } @@ -540,6 +578,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 @@ -564,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; @@ -574,8 +643,8 @@ 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}; - use http::{HttpResult, HttpScheme, Header, OwnedHeader, ErrorCode}; + WindowUpdateFrame, PingFrame, pack_header, RawFrame, FrameIR}; + use http::{HttpResult, HttpError, HttpScheme, Header, OwnedHeader, ErrorCode}; use hpack; /// A helper function that performs a `send_frame` operation on the given @@ -890,6 +959,71 @@ 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); + } + + #[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] @@ -924,13 +1058,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 @@ -943,10 +1085,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], @@ -957,6 +1097,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); @@ -1000,6 +1144,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/flow_control.rs b/src/http/flow_control.rs new file mode 100644 index 00000000..e554e412 --- /dev/null +++ b/src/http/flow_control.rs @@ -0,0 +1,129 @@ +//! 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. + 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. + 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 1fed8257..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; @@ -522,6 +523,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 +531,37 @@ 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() + } +} + +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. #[derive(Debug, Copy, Clone, PartialEq)] pub enum HttpScheme { 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 0686b4cd..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; @@ -78,25 +79,98 @@ 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(()) + } + + /// 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`. /// /// 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, + /// 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 { + /// Create a new `Entry` with the given `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 + 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 } + + 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. /// /// This trait is tightly coupled to a `Stream`-based session layer implementation. Particular @@ -118,15 +192,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; @@ -134,22 +206,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())) } } @@ -188,7 +271,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 @@ -260,7 +343,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 } @@ -268,24 +351,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) } @@ -355,9 +437,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); @@ -370,6 +454,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 @@ -393,6 +485,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. @@ -403,6 +496,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. @@ -413,12 +507,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() { @@ -426,6 +522,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() { @@ -542,7 +639,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}; @@ -756,6 +853,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. diff --git a/src/http/tests/common.rs b/src/http/tests/common.rs index 20952257..758cb282 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,14 @@ 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)>, + /// 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 { @@ -268,6 +276,10 @@ impl TestSession { goaways: Vec::new(), pings: Vec::new(), pongs: Vec::new(), + conn_window_updates: Vec::new(), + stream_window_updates: Vec::new(), + conn_in_window_decreases: Vec::new(), + stream_window_decreases: Vec::new(), } } @@ -285,6 +297,10 @@ impl TestSession { goaways: Vec::new(), pings: Vec::new(), pongs: Vec::new(), + conn_window_updates: Vec::new(), + stream_window_updates: Vec::new(), + conn_in_window_decreases: Vec::new(), + stream_window_decreases: Vec::new(), } } } @@ -353,6 +369,35 @@ 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(()) + } + + 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. 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,