// This is used for filtering send_streams and recv_Streams with a stream_ids greater than or equal // a given id. Only the same type (bidirectional or unidirectionsl) streams are filtered. fn id_gte<U>(base: StreamId) -> impl FnMut((&StreamId, &U)) -> Option<StreamId> + 'static where
U: ?Sized,
{ move |(id, _)| { if *id >= base && !(id.is_bidi() ^ base.is_bidi()) {
Some(*id)
} else {
None
}
}
}
impl Http3Client { /// # Errors /// /// Making a `neqo-transport::connection` may produce an error. This can only be a crypto error /// if the crypto context can't be created or configured. pubfn new(
server_name: impl Into<String>,
cid_manager: Rc<RefCell<dyn ConnectionIdGenerator>>,
local_addr: SocketAddr,
remote_addr: SocketAddr,
http3_parameters: Http3Parameters,
now: Instant,
) -> Res<Self> {
Ok(Self::new_with_conn(
Connection::new_client(
server_name,
&[alpn_from_quic_version(
http3_parameters
.get_connection_parameters()
.get_versions()
.initial(),
)],
cid_manager,
local_addr,
remote_addr,
http3_parameters.get_connection_parameters().clone(),
now,
)?,
http3_parameters,
))
}
/// This is a similar function to `new`. In this case, `neqo-transport::connection` has been /// already created. /// /// It is recommended to use `new` instead. #[must_use] pubfn new_with_conn(c: Connection, http3_parameters: Http3Parameters) -> Self { let events = Http3ClientEvents::default(); let webtransport = http3_parameters.get_webtransport(); let push_streams = http3_parameters.get_max_concurrent_push_streams(); letmut base_handler = Http3Connection::new(http3_parameters, Role::Client); if webtransport {
base_handler.set_features_listener(events.clone());
} Self {
conn: c,
events: events.clone(),
push_handler: Rc::new(RefCell::new(PushController::new(push_streams, events))),
base_handler,
}
}
#[must_use] pubconstfn role(&self) -> Role { self.conn.role()
}
/// The function returns the current state of the connection. #[must_use] pubfn state(&self) -> Http3State { self.base_handler.state()
}
/// Get the peer's certificate. #[must_use] pubfn peer_certificate(&self) -> Option<CertificateInfo> { self.conn.peer_certificate()
}
/// This called when peer certificates have been verified. /// /// `Http3ClientEvent::AuthenticationNeeded` event is emitted when peer’s certificates are /// available and need to be verified. When the verification is completed this function is /// called. To inform HTTP/3 session of the verification results. pubfn authenticated(&mutself, status: AuthenticationStatus, now: Instant) { self.conn.authenticated(status, now);
}
/// Enable encrypted client hello (ECH). /// /// # Errors /// /// Fails when the configuration provided is bad. pubfn enable_ech(&mutself, ech_config_list: impl AsRef<[u8]>) -> Res<()> { self.conn.client_enable_ech(ech_config_list)?;
Ok(())
}
/// Get the connection id, which is useful for disambiguating connections to /// the same origin. /// /// # Panics /// /// Never, because clients always have this field. #[must_use] pubfn connection_id(&self) -> &ConnectionId { self.conn.odcid().expect("Client always has odcid")
}
/// The correct way to obtain a resumption token is to wait for the /// `Http3ClientEvent::ResumptionToken` event. To emit the event we are waiting for a /// resumtion token and a `NEW_TOKEN` frame to arrive. Some servers don't send `NEW_TOKEN` /// frames and in this case, we wait for 3xPTO before emitting an event. This is especially a /// problem for short-lived connections, where the connection is closed before any events are /// released. This function retrieves the token, without waiting for a `NEW_TOKEN` frame to /// arrive. /// /// In addition to the token, HTTP/3 settings are encoded into the token before giving it to /// the application(`encode_resumption_token`). When the resumption token is supplied to a new /// connection the HTTP/3 setting will be decoded and used until the setting are received from /// the server. pubfn take_resumption_token(&mutself, now: Instant) -> Option<ResumptionToken> { self.conn
.take_resumption_token(now)
.and_then(|t| self.encode_resumption_token(&t))
}
/// This may be call if an application has a resumption token. This must be called before /// connection starts. /// /// The resumption token also contains encoded HTTP/3 settings. The settings will be decoded /// and used until the setting are received from the server. /// /// # Errors /// /// An error is return if token cannot be decoded or a connection is is a wrong state. /// /// # Panics /// /// On closing if the base handler can't handle it (debug only). pubfn enable_resumption(&mutself, now: Instant, token: impl AsRef<[u8]>) -> Res<()> { ifself.base_handler.state != Http3State::Initializing { return Err(Error::InvalidState);
} letmut dec = Decoder::from(token.as_ref()); let Some(settings_slice) = dec.decode_vvec() else { return Err(Error::InvalidResumptionToken);
};
qtrace!([self], " settings {}", hex_with_len(settings_slice)); letmut dec_settings = Decoder::from(settings_slice); letmut settings = HSettings::default();
Error::map_error(
settings.decode_frame_contents(&mut dec_settings),
Error::InvalidResumptionToken,
)?; let tok = dec.decode_remainder();
qtrace!([self], " Transport token {}", hex(tok)); self.conn.enable_resumption(now, tok)?; ifself.conn.state().closed() { let state = self.conn.state().clone(); let res = self
.base_handler
.handle_state_change(&mutself.conn, &state);
debug_assert_eq!(Ok(true), res); return Err(Error::FatalError);
} ifself.conn.zero_rtt_state() == ZeroRttState::Sending { self.base_handler
.set_0rtt_settings(&mutself.conn, settings)?; self.events
.connection_state_change(self.base_handler.state()); self.push_handler
.borrow_mut()
.maybe_send_max_push_id_frame(&mutself.base_handler);
}
Ok(())
}
/// This is call to close a connection. pubfn close<S>(&mutself, now: Instant, error: AppError, msg: S) where
S: AsRef<str> + Display,
{
qinfo!([self], "Close the connection error={} msg={}.", error, msg); if !matches!( self.base_handler.state,
Http3State::Closing(_) | Http3State::Closed(_)
) { self.push_handler.borrow_mut().clear(); self.conn.close(now, error, msg); self.base_handler.close(error); self.events
.connection_state_change(self.base_handler.state());
}
}
/// Attempt to force a key update. /// /// # Errors /// /// If the connection isn't confirmed, or there is an outstanding key update, this /// returns `Err(Error::TransportError(neqo_transport::Error::KeyUpdateBlocked))`. pubfn initiate_key_update(&mutself) -> Res<()> { self.conn.initiate_key_update()?;
Ok(())
}
// API: Request/response
/// The function fetches a resource using `method`, `target` and `headers`. A response body /// may be added by calling `send_data`. `stream_close_send` must be sent to finish the request /// even if request data are not sent. /// /// # Errors /// /// If a new stream cannot be created an error will be return. /// /// # Panics /// /// `SendMessage` implements `http_stream` so it will not panic. pubfn fetch<'x, 't: 'x, T>(
&mutself,
now: Instant,
method: &'t str,
target: &'t T,
headers: &'t [Header],
priority: Priority,
) -> Res<StreamId> where
T: AsRequestTarget<'x> + ?Sized + Debug,
{ let output = self.base_handler.fetch(
&mutself.conn, Box::new(self.events.clone()), Box::new(self.events.clone()),
Some(Rc::clone(&self.push_handler)),
&RequestDescription {
method,
connect_type: None,
target,
headers,
priority,
},
); iflet Err(e) = &output { if e.connection_error() { self.close(now, e.code(), "");
}
}
output
}
/// Send an [`PRIORITY_UPDATE`-frame][1] on next `Http3Client::process_output()` call. /// Returns if the priority got changed. /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist /// /// [1]: https://datatracker.ietf.org/doc/html/draft-kazuho-httpbis-priority-04#section-5.2 pubfn priority_update(&mutself, stream_id: StreamId, priority: Priority) -> Res<bool> { self.base_handler.queue_update_priority(stream_id, priority)
}
/// An application may cancel a stream(request). /// Both sides, the receiviing and sending side, sending and receiving side, will be closed. /// /// # Errors /// /// An error will be return if a stream does not exist. pubfn cancel_fetch(&mutself, stream_id: StreamId, error: AppError) -> Res<()> {
qinfo!([self], "reset_stream {} error={}.", stream_id, error); self.base_handler
.cancel_fetch(stream_id, error, &mutself.conn)
}
/// This is call when application is done sending a request. /// /// # Errors /// /// An error will be return if stream does not exist. pubfn stream_close_send(&mutself, stream_id: StreamId) -> Res<()> {
qdebug!([self], "Close sending side stream={}.", stream_id); self.base_handler
.stream_close_send(&mutself.conn, stream_id)
}
/// # Errors /// /// An error will be return if a stream does not exist. pubfn stream_reset_send(&mutself, stream_id: StreamId, error: AppError) -> Res<()> {
qinfo!([self], "stream_reset_send {} error={}.", stream_id, error); self.base_handler
.stream_reset_send(&mutself.conn, stream_id, error)
}
/// # Errors /// /// An error will be return if a stream does not exist. pubfn stream_stop_sending(&mutself, stream_id: StreamId, error: AppError) -> Res<()> {
qinfo!([self], "stream_stop_sending {} error={}.", stream_id, error); self.base_handler
.stream_stop_sending(&mutself.conn, stream_id, error)
}
/// This function is used for regular HTTP requests and `WebTransport` streams. /// In the case of regular HTTP requests, the request body is supplied using this function, and /// headers are supplied through the `fetch` function. /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist, /// `AlreadyClosed` if the stream has already been closed. /// `TransportStreamDoesNotExist` if the transport stream does not exist (this may happen if /// `process_output` has not been called when needed, and HTTP3 layer has not picked up the /// info that the stream has been closed.) `InvalidInput` if an empty buffer has been /// supplied. pubfn send_data(&mutself, stream_id: StreamId, buf: &[u8]) -> Res<usize> {
qinfo!(
[self], "send_data from stream {} sending {} bytes.",
stream_id,
buf.len()
); self.base_handler
.send_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?
.send_data(&mutself.conn, buf)
}
/// Response data are read directly into a buffer supplied as a parameter of this function to /// avoid copying data. /// /// # Errors /// /// It returns an error if a stream does not exist or an error happen while reading a stream, /// e.g. early close, protocol error, etc. pubfn read_data(
&mutself,
now: Instant,
stream_id: StreamId,
buf: &mut [u8],
) -> Res<(usize, bool)> {
qdebug!([self], "read_data from stream {}.", stream_id); let res = self.base_handler.read_data(&mutself.conn, stream_id, buf); iflet Err(e) = &res { if e.connection_error() { self.close(now, e.code(), "");
}
}
res
}
// API: Push streams
/// Cancel a push /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist. pubfn cancel_push(&mutself, push_id: u64) -> Res<()> { self.push_handler
.borrow_mut()
.cancel(push_id, &mutself.conn, &mutself.base_handler)
}
/// Push response data are read directly into a buffer supplied as a parameter of this function /// to avoid copying data. /// /// # Errors /// /// It returns an error if a stream does not exist(`InvalidStreamId`) or an error has happened /// while reading a stream, e.g. early close, protocol error, etc. pubfn push_read_data(
&mutself,
now: Instant,
push_id: u64,
buf: &mut [u8],
) -> Res<(usize, bool)> { let stream_id = self
.push_handler
.borrow_mut()
.get_active_stream_id(push_id)
.ok_or(Error::InvalidStreamId)?; self.conn.stream_keep_alive(stream_id, true)?; self.read_data(now, stream_id, buf)
}
// API WebTransport // /// # Errors /// /// If `WebTransport` cannot be created, e.g. the `WebTransport` support is /// not negotiated or the HTTP/3 connection is closed. pubfn webtransport_create_session<'x, 't: 'x, T>(
&mutself,
now: Instant,
target: &'t T,
headers: &'t [Header],
) -> Res<StreamId> where
T: AsRequestTarget<'x> + ?Sized + Debug,
{ let output = self.base_handler.webtransport_create_session(
&mutself.conn, Box::new(self.events.clone()),
target,
headers,
);
/// Close `WebTransport` cleanly /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist, /// `TransportStreamDoesNotExist` if the transport stream does not exist (this may happen if /// `process_output` has not been called when needed, and HTTP3 layer has not picked up the /// info that the stream has been closed.) `InvalidInput` if an empty buffer has been /// supplied. pubfn webtransport_close_session(
&mutself,
session_id: StreamId,
error: u32,
message: &str,
) -> Res<()> { self.base_handler
.webtransport_close_session(&mutself.conn, session_id, error, message)
}
/// # Errors /// /// This may return an error if the particular session does not exist /// or the connection is not in the active state. pubfn webtransport_create_stream(
&mutself,
session_id: StreamId,
stream_type: StreamType,
) -> Res<StreamId> { self.base_handler.webtransport_create_stream_local(
&mutself.conn,
session_id,
stream_type, Box::new(self.events.clone()), Box::new(self.events.clone()),
)
}
/// Send `WebTransport` datagram. /// /// # Errors /// /// It may return `InvalidStreamId` if a stream does not exist anymore. /// The function returns `TooMuchData` if the supply buffer is bigger than /// the allowed remote datagram size. pubfn webtransport_send_datagram(
&mutself,
session_id: StreamId,
buf: &[u8],
id: impl Into<DatagramTracking>,
) -> Res<()> {
qtrace!("webtransport_send_datagram session:{:?}", session_id); self.base_handler
.webtransport_send_datagram(session_id, &mutself.conn, buf, id)
}
/// Returns the current max size of a datagram that can fit into a packet. /// The value will change over time depending on the encoded size of the /// packet number, ack frames, etc. /// /// # Errors /// /// The function returns `NotAvailable` if datagrams are not enabled. /// /// # Panics /// /// This cannot panic. The max varint length is 8. pubfn webtransport_max_datagram_size(&self, session_id: StreamId) -> Res<u64> {
Ok(self.conn.max_datagram_size()?
- u64::try_from(Encoder::varint_len(session_id.as_u64())).unwrap())
}
/// Sets the `SendOrder` for a given stream /// /// # Errors /// /// It may return `InvalidStreamId` if a stream does not exist anymore. /// /// # Panics /// /// This cannot panic. pubfn webtransport_set_sendorder(
&mutself,
stream_id: StreamId,
sendorder: Option<SendOrder>,
) -> Res<()> {
Http3Connection::stream_set_sendorder(&mutself.conn, stream_id, sendorder)
}
/// Sets the `Fairness` for a given stream /// /// # Errors /// /// It may return `InvalidStreamId` if a stream does not exist anymore. /// /// # Panics /// /// This cannot panic. pubfn webtransport_set_fairness(&mutself, stream_id: StreamId, fairness: bool) -> Res<()> {
Http3Connection::stream_set_fairness(&mutself.conn, stream_id, fairness)
}
/// Returns the current `SendStreamStats` of a `WebTransportSendStream`. /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist. pubfn webtransport_send_stream_stats(&mutself, stream_id: StreamId) -> Res<SendStreamStats> { self.base_handler
.send_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?
.stats(&mutself.conn)
}
/// Returns the current `RecvStreamStats` of a `WebTransportRecvStream`. /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist. pubfn webtransport_recv_stream_stats(&mutself, stream_id: StreamId) -> Res<RecvStreamStats> { self.base_handler
.recv_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?
.stats(&mutself.conn)
}
/// This function combines `process_input` and `process_output` function. pubfn process(&mutself, dgram: Option<Datagram<impl AsRef<[u8]>>>, now: Instant) -> Output {
qtrace!([self], "Process."); iflet Some(d) = dgram { self.process_input(d, now);
} self.process_output(now)
}
/// The function should be called when there is a new UDP packet available. The function will /// handle the packet payload. /// /// First, the payload will be handled by the QUIC layer. Afterward, `process_http3` will be /// called to handle new [`ConnectionEvent`][1]s. /// /// After this function is called `process_output` should be called to check whether new /// packets need to be sent or if a timer needs to be updated. /// /// [1]: ../neqo_transport/enum.ConnectionEvent.html pubfn process_input(&mutself, dgram: Datagram<impl AsRef<[u8]>>, now: Instant) { self.process_multiple_input(iter::once(dgram), now);
}
/// Process HTTP3 layer. /// When `process_output`, `process_input`, or `process` is called we must call this function /// as well. The functions calls `Http3Client::check_connection_events` to handle events from /// the QUC layer and calls `Http3Connection::process_sending` to ensure that HTTP/3 layer /// data, e.g. control frames, are sent. fn process_http3(&mutself, now: Instant) {
qtrace!([self], "Process http3 internal."); matchself.base_handler.state() {
Http3State::ZeroRtt | Http3State::Connected | Http3State::GoingAway(..) => { let res = self.check_connection_events(); ifself.check_result(now, &res) { return;
} self.push_handler
.borrow_mut()
.maybe_send_max_push_id_frame(&mutself.base_handler); let res = self.base_handler.process_sending(&mutself.conn); self.check_result(now, &res);
}
Http3State::Closed { .. } => {}
_ => { let res = self.check_connection_events();
_ = self.check_result(now, &res);
}
}
}
/// The function should be called to check if there is a new UDP packet to be sent. It should /// be called after a new packet is received and processed and after a timer expires (QUIC /// needs timers to handle events like PTO detection and timers are not implemented by the neqo /// library, but instead must be driven by the application). /// /// `process_output` can return: /// - a [`Output::Datagram(Datagram)`][1]: data that should be sent as a UDP payload, /// - a [`Output::Callback(Duration)`][1]: the duration of a timer. `process_output` should be /// called at least after the time expires, /// - [`Output::None`][1]: this is returned when `Nttp3Client` is done and can be destroyed. /// /// The application should call this function repeatedly until a timer value or None is /// returned. After that, the application should call the function again if a new UDP packet is /// received and processed or the timer value expires. /// /// The HTTP/3 neqo implementation drives the HTTP/3 and QUIC layers, therefore this function /// will call both layers: /// - First it calls HTTP/3 layer processing (`process_http3`) to make sure the layer writes /// data to QUIC layer or cancels streams if needed. /// - Then QUIC layer processing is called - [`Connection::process_output`][3]. This produces a /// packet or a timer value. It may also produce new [`ConnectionEvent`][2]s, e.g. connection /// state-change event. /// - Therefore the HTTP/3 layer processing (`process_http3`) is called again. /// /// [1]: ../neqo_transport/enum.Output.html /// [2]: ../neqo_transport/struct.ConnectionEvents.html /// [3]: ../neqo_transport/struct.Connection.html#method.process_output pubfn process_output(&mutself, now: Instant) -> Output {
qtrace!([self], "Process output.");
// Maybe send() stuff on http3-managed streams self.process_http3(now);
let out = self.conn.process_output(now);
// Update H3 for any transport state changes and events self.process_http3(now);
out
}
/// This function takes the provided result and check for an error. /// An error results in closing the connection. fn check_result<ERR>(&mutself, now: Instant, res: &Res<ERR>) -> bool { match &res {
Err(Error::HttpGoaway) => {
qinfo!([self], "Connection error: goaway stream_id increased."); self.close(
now,
Error::HttpGeneralProtocol.code(), "Connection error: goaway stream_id increased",
); true
}
Err(e) => {
qinfo!([self], "Connection error: {}.", e); self.close(now, e.code(), format!("{e}")); true
}
_ => false,
}
}
/// This function checks [`ConnectionEvent`][2]s emitted by the QUIC layer, e.g. connection /// change state events, new incoming stream data is available, a stream is was reset, etc. /// The HTTP/3 layer needs to handle these events. Most of the events are handled by /// [`Http3Connection`][1] by calling appropriate functions, e.g. `handle_state_change`, /// `handle_stream_reset`, etc. [`Http3Connection`][1] handle functionalities that are common /// for the client and server side. Some of the functionalities are specific to the client and /// they are handled by `Http3Client`. For example, [`ConnectionEvent::RecvStreamReadable`][3] /// event is handled by `Http3Client::handle_stream_readable`. The function calls /// `Http3Connection::handle_stream_readable` and then hands the return value as appropriate /// for the client-side. /// /// [1]: https://github.com/mozilla/neqo/blob/main/neqo-http3/src/connection.rs /// [2]: ../neqo_transport/enum.ConnectionEvent.html /// [3]: ../neqo_transport/enum.ConnectionEvent.html#variant.RecvStreamReadable fn check_connection_events(&mutself) -> Res<()> {
qtrace!([self], "Check connection events."); whilelet Some(e) = self.conn.next_event() {
qdebug!([self], "check_connection_events - event {:?}.", e); match e {
ConnectionEvent::NewStream { stream_id } => { // During this event we only add a new stream to the Http3Connection stream // list, with NewStreamHeadReader stream handler. // This function will not read from the stream and try to decode the stream. // RecvStreamReadable will be emitted after this event and reading, i.e. // decoding of a stream will happen during that event. self.base_handler.add_new_stream(stream_id);
}
ConnectionEvent::SendStreamWritable { stream_id } => { iflet Some(s) = self.base_handler.send_streams.get_mut(&stream_id) {
s.stream_writable();
}
}
ConnectionEvent::RecvStreamReadable { stream_id } => { self.handle_stream_readable(stream_id)?;
}
ConnectionEvent::RecvStreamReset {
stream_id,
app_error,
} => self
.base_handler
.handle_stream_reset(stream_id, app_error, &mutself.conn)?,
ConnectionEvent::SendStreamStopSending {
stream_id,
app_error,
} => self.base_handler.handle_stream_stop_sending(
stream_id,
app_error,
&mutself.conn,
)?,
/// This function handled new data available on a stream. It calls /// `Http3Client::handle_stream_readable` and handles its response. Reading streams are mostly /// handled by [`Http3Connection`][1] because most part of it is common for the client and /// server. The following actions need to be handled by the client-specific code: /// - `ReceiveOutput::NewStream(NewStreamType::Push(_))` - the server cannot receive a push /// stream, /// - `ReceiveOutput::NewStream(NewStreamType::Http)` - client cannot receive a /// server-initiated HTTP request, /// - `ReceiveOutput::NewStream(NewStreamType::WebTransportStream(_))` - because /// `Http3ClientEvents`is needed and events handler is specific to the client. /// - `ReceiveOutput::ControlFrames(control_frames)` - some control frame handling differs /// between the client and the server: /// - `HFrame::CancelPush` - only the client-side may receive it, /// - `HFrame::MaxPushId { .. }`, `HFrame::PriorityUpdateRequest { .. } ` and /// `HFrame::PriorityUpdatePush` can only be receive on the server side, /// - `HFrame::Goaway { stream_id }` needs specific handling by the client by the protocol /// specification. /// /// [1]: https://github.com/mozilla/neqo/blob/main/neqo-http3/src/connection.rs fn handle_stream_readable(&mutself, stream_id: StreamId) -> Res<()> { matchself
.base_handler
.handle_stream_readable(&mutself.conn, stream_id)?
{
ReceiveOutput::NewStream(NewStreamType::Push(push_id)) => { self.handle_new_push_stream(stream_id, push_id)
}
ReceiveOutput::NewStream(NewStreamType::Http(_)) => Err(Error::HttpStreamCreation),
ReceiveOutput::NewStream(NewStreamType::WebTransportStream(session_id)) => { self.base_handler.webtransport_create_stream_remote(
StreamId::from(session_id),
stream_id, Box::new(self.events.clone()), Box::new(self.events.clone()),
)?; let res = self
.base_handler
.handle_stream_readable(&mutself.conn, stream_id)?;
debug_assert!(matches!(res, ReceiveOutput::NoOutput));
Ok(())
}
ReceiveOutput::ControlFrames(control_frames) => { for f in control_frames { match f {
HFrame::CancelPush { push_id } => self
.push_handler
.borrow_mut()
.handle_cancel_push(push_id, &mutself.conn, &mutself.base_handler),
HFrame::MaxPushId { .. }
| HFrame::PriorityUpdateRequest { .. }
| HFrame::PriorityUpdatePush { .. } => Err(Error::HttpFrameUnexpected),
HFrame::Goaway { stream_id } => self.handle_goaway(stream_id),
_ => {
unreachable!( "we should only put MaxPushId, Goaway and PriorityUpdates into control_frames."
);
}
}?;
}
Ok(())
}
_ => Ok(()),
}
}
// Add a new push stream to `PushController`. `add_new_push_stream` may return an error // (this will be a connection error) or a bool. // If false is returned that means that the stream should be reset because the push has // been already canceled (CANCEL_PUSH frame or canceling push from the application). if !self
.push_handler
.borrow_mut()
.add_new_push_stream(push_id, stream_id)?
{ // We are not interested in the result of stream_stop_sending, we are not interested // in this stream.
mem::drop( self.conn
.stream_stop_sending(stream_id, Error::HttpRequestCancelled.code()),
); return Ok(());
}
self.base_handler.add_recv_stream(
stream_id, Box::new(RecvMessage::new(
&RecvMessageInfo {
message_type: MessageType::Response,
stream_type: Http3StreamType::Push,
stream_id,
first_frame_type: None,
},
Rc::clone(&self.base_handler.qpack_decoder), Box::new(RecvPushEvents::new(push_id, Rc::clone(&self.push_handler))),
None, // TODO: think about the right prority for the push streams.
PriorityHandler::new(true, Priority::default()),
)),
); let res = self
.base_handler
.handle_stream_readable(&mutself.conn, stream_id)?;
debug_assert!(matches!(res, ReceiveOutput::NoOutput));
Ok(())
}
// Issue reset events for streams >= goaway stream id let send_ids: Vec<StreamId> = self
.base_handler
.send_streams
.iter()
.filter_map(id_gte(goaway_stream_id))
.collect(); for id in send_ids { // We do not care about streams that are going to be closed.
mem::drop(self.base_handler.handle_stream_stop_sending(
id,
Error::HttpRequestRejected.code(),
&mutself.conn,
));
}
let recv_ids: Vec<StreamId> = self
.base_handler
.recv_streams
.iter()
.filter_map(id_gte(goaway_stream_id))
.collect(); for id in recv_ids { // We do not care about streams that are going to be closed.
mem::drop(self.base_handler.handle_stream_reset(
id,
Error::HttpRequestRejected.code(),
&mutself.conn,
));
}
self.events.goaway_received();
Ok(())
}
/// Increases `max_stream_data` for a `stream_id`. /// /// # Errors /// /// Returns `InvalidStreamId` if a stream does not exist or the receiving /// side is closed. pubfn set_stream_max_data(&mutself, stream_id: StreamId, max_data: u64) -> Res<()> { self.conn.set_stream_max_data(stream_id, max_data)?;
Ok(())
}
impl EventProvider for Http3Client { type Event = Http3ClientEvent;
/// Return true if there are outstanding events. fn has_events(&self) -> bool { self.events.has_events()
}
/// Get events that indicate state changes on the connection. This method /// correctly handles cases where handling one event can obsolete /// previously-queued events, or cause new events to be generated. fn next_event(&mutself) -> Option<Self::Event> { self.events.next_event()
}
}
#[cfg(test)] mod tests { use std::{mem, time::Duration};
use neqo_common::{event::Provider, qtrace, Datagram, Decoder, Encoder}; use neqo_crypto::{AllowZeroRtt, AntiReplay, ResumptionToken}; use neqo_qpack::{encoder::QPackEncoder, QpackSettings}; use neqo_transport::{
CloseReason, ConnectionEvent, ConnectionParameters, Output, State, StreamId, StreamType,
Version, MIN_INITIAL_PACKET_SIZE, RECV_BUFFER_SIZE, SEND_BUFFER_SIZE,
}; use test_fixture::{
anti_replay, default_server_h3, fixture_init, new_server, now,
CountingConnectionIdGenerator, DEFAULT_ADDR, DEFAULT_ALPN_H3, DEFAULT_KEYS,
DEFAULT_SERVER_NAME,
};
// Encoder stream data with a change capacity instruction(0x3f, 0x45 = change capacity to 100) // This data will be send when 0-RTT is used and we already have a max_table_capacity from // resumed settings. const ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION: &[u8] = &[0x2, 0x3f, 0x45];
pubfn create_control_stream(&mutself) { // Create control stream let control = self.conn.stream_create(StreamType::UniDi).unwrap();
qtrace!(["TestServer"], "control stream: {}", control); self.control_stream_id = Some(control); // Send stream type on the control stream.
assert_eq!( self.conn
.stream_send(self.control_stream_id.unwrap(), CONTROL_STREAM_TYPE)
.unwrap(), 1
);
// Encode a settings frame and send it. letmut enc = Encoder::default(); self.settings.encode(&mut enc);
assert_eq!( self.conn
.stream_send(self.control_stream_id.unwrap(), enc.as_ref())
.unwrap(),
enc.len()
);
}
fn handshake_only(client: &mut Http3Client, server: &'color:red'>mut TestServer) -> Output {
assert_eq!(client.state(), Http3State::Initializing); let out = client.process_output(now());
assert_eq!(client.state(), Http3State::Initializing);
assert_eq!(*server.conn.state(), State::Init); let out = server.conn.process(out.dgram(), now());
assert_eq!(*server.conn.state(), State::Handshaking);
let out = client.process(out.dgram(), now()); let out = server.conn.process(out.dgram(), now());
assert!(out.as_dgram_ref().is_none());
let authentication_needed = |e| matches!(e, Http3ClientEvent::AuthenticationNeeded);
assert!(client.events().any(authentication_needed));
client.authenticated(AuthenticationStatus::Ok, now());
out
}
// Perform only Quic transport handshake. fn connect_only_transport_with(client: &mut Http3Client, server: &mut TestServer) { let out = handshake_only(client, server);
let out = client.process(out.dgram(), now()); let connected = |e| matches!(e, Http3ClientEvent::StateChange(Http3State::Connected));
assert!(client.events().any(connected));
// Perform only Quic transport handshake. fn connect_only_transport() -> (Http3Client, TestServer) { letmut client = default_http3_client(); letmut server = TestServer::new();
connect_only_transport_with(&mut client, &mut server);
(client, server)
}
fn send_and_receive_client_settings(client: &mut Http3Client, server: &mut TestServer) { // send and receive client settings let out = client.process_output(now());
server.conn.process_input(out.dgram().unwrap(), now());
server.check_client_control_qpack_streams_no_resumption();
}
// Perform Quic transport handshake and exchange Http3 settings. fn connect_with(client: &mut Http3Client, server: &mut TestServer) {
connect_only_transport_with(client, server);
send_and_receive_client_settings(client, server);
server.create_control_stream();
server.create_qpack_streams(); // Send the server's control and qpack streams data. let out = server.conn.process(None::<Datagram>, now());
client.process_input(out.dgram().unwrap(), now());
// assert no error occured.
assert_eq!(client.state(), Http3State::Connected);
}
// Perform Quic transport handshake and exchange Http3 settings. fn connect_with_connection_parameters(
server_conn_params: ConnectionParameters,
) -> (Http3Client, TestServer) { // connecting with default max_table_size letmut client = default_http3_client_param(100); let server = Connection::new_server(
test_fixture::DEFAULT_KEYS,
test_fixture::DEFAULT_ALPN_H3,
Rc::new(RefCell::new(CountingConnectionIdGenerator::default())),
server_conn_params,
)
.unwrap(); letmut server = TestServer::new_with_conn(server);
connect_with(&mut client, &mut server);
(client, server)
}
// Perform Quic transport handshake and exchange Http3 settings. fn connect() -> (Http3Client, TestServer) { letmut client = default_http3_client(); letmut server = TestServer::new();
connect_with(&mut client, &mut server);
(client, server)
}
// The response header from PUSH_DATA (0x01, 0x06, 0x00, 0x00, 0xd9, 0x54, 0x01, 0x34) are // decoded into: fn check_push_response_header(header: &[Header]) { let expected_push_response_header = [
Header::new(":status", "200"),
Header::new("content-length", "4"),
];
assert_eq!(header, &expected_push_response_header[..]);
}
// The data frame payload from PUSH_DATA is: const EXPECTED_PUSH_RESPONSE_DATA_FRAME: &[u8] = &[0x61, 0x62, 0x63, 0x64];
// Send push data on a push stream: // 1) push_stream_type PUSH_STREAM_TYPE // 2) push_id // 3) PUSH_DATA that contains encoded headers and a data frame. // This function can only handle small push_id numbers that fit in a varint of length 1 byte. fn send_data_on_push(
conn: &mut Connection,
push_stream_id: StreamId,
push_id: u8,
data: impl AsRef<[u8]>,
close_push_stream: bool,
) { // send data
_ = conn.stream_send(push_stream_id, PUSH_STREAM_TYPE).unwrap();
_ = conn.stream_send(push_stream_id, &[push_id]).unwrap();
_ = conn.stream_send(push_stream_id, data.as_ref()).unwrap(); if close_push_stream {
conn.stream_close_send(push_stream_id).unwrap();
}
}
// Send push data on a push stream: // 1) push_stream_type PUSH_STREAM_TYPE // 2) push_id // 3) PUSH_DATA that contains encoded headers and a data frame. // This function can only handle small push_id numbers that fit in a varint of length 1 byte. fn send_push_data(conn: &mut Connection, push_id: u8, close_push_stream: bool) -> StreamId {
send_push_with_data(conn, push_id, PUSH_DATA, close_push_stream)
}
// Send push data on a push stream: // 1) push_stream_type PUSH_STREAM_TYPE // 2) push_id // 3) and supplied push data. // This function can only handle small push_id numbers that fit in a varint of length 1 byte. fn send_push_with_data(
conn: &mut Connection,
push_id: u8,
data: &[u8],
close_push_stream: bool,
) -> StreamId { // create a push stream let push_stream_id = conn.stream_create(StreamType::UniDi).unwrap(); // send data
send_data_on_push(conn, push_stream_id, push_id, data, close_push_stream);
push_stream_id
}
// Client: Test receiving a new control stream and a SETTINGS frame. #[test] fn client_connect_and_exchange_qpack_and_control_streams() {
mem::drop(connect());
}
// Client: Test that the connection will be closed if control stream // has been closed. #[test] fn client_close_control_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_close_send(server.control_stream_id.unwrap())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the local control stream // has been reset. #[test] fn client_reset_control_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_reset_send(server.control_stream_id.unwrap(), Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the server side encoder stream // has been reset. #[test] fn client_reset_server_side_encoder_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_reset_send(server.encoder_stream_id.unwrap(), Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the server side decoder stream // has been reset. #[test] fn client_reset_server_side_decoder_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_reset_send(server.decoder_stream_id.unwrap(), Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the local control stream // has received a stop_sending. #[test] fn client_stop_sending_control_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_stop_sending(CLIENT_SIDE_CONTROL_STREAM_ID, Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the client side encoder stream // has received a stop_sending. #[test] fn client_stop_sending_encoder_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_stop_sending(CLIENT_SIDE_ENCODER_STREAM_ID, Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: Test that the connection will be closed if the client side decoder stream // has received a stop_sending. #[test] fn client_stop_sending_decoder_stream() { let (mut client, mut server) = connect();
server
.conn
.stream_stop_sending(CLIENT_SIDE_DECODER_STREAM_ID, Error::HttpNoError.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpClosedCriticalStream);
}
// Client: test missing SETTINGS frame // (the first frame sent is a garbage frame). #[test] fn client_missing_settings() { let (mut client, mut server) = connect_only_transport(); // Create server control stream. let control_stream = server.conn.stream_create(StreamType::UniDi).unwrap(); // Send a HEADERS frame instead (which contains garbage). let sent = server
.conn
.stream_send(control_stream, &[0x0, 0x1, 0x3, 0x0, 0x1, 0x2]);
assert_eq!(sent, Ok(6)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpMissingSettings);
}
// Client: receiving SETTINGS frame twice causes connection close // with error HTTP_UNEXPECTED_FRAME. #[test] fn client_receive_settings_twice() { let (mut client, mut server) = connect(); // send the second SETTINGS frame. let sent = server.conn.stream_send(
server.control_stream_id.unwrap(),
&[0x4, 0x6, 0x1, 0x40, 0x64, 0x7, 0x40, 0x64],
);
assert_eq!(sent, Ok(8)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpFrameUnexpected);
}
// send DATA frame on a cortrol stream #[test] fn data_frame_on_control_stream() {
test_wrong_frame_on_control_stream(&[0x0, 0x2, 0x1, 0x2]);
}
// send HEADERS frame on a cortrol stream #[test] fn headers_frame_on_control_stream() {
test_wrong_frame_on_control_stream(&[0x1, 0x2, 0x1, 0x2]);
}
// send PUSH_PROMISE frame on a cortrol stream #[test] fn push_promise_frame_on_control_stream() {
test_wrong_frame_on_control_stream(&[0x5, 0x2, 0x1, 0x2]);
}
// send PRIORITY_UPDATE frame on a control stream to the client #[test] fn priority_update_request_on_control_stream() {
test_wrong_frame_on_control_stream(&[0x80, 0x0f, 0x07, 0x00, 0x01, 0x03]);
}
send_push_promise(&mut server.conn, request_stream_id, 0); // Create a push stream let push_stream_id = server.conn.stream_create(StreamType::UniDi).unwrap();
// Send the push stream type byte, push_id and frame v.
_ = server
.conn
.stream_send(push_stream_id, &[0x01, 0x0])
.unwrap();
_ = server.conn.stream_send(push_stream_id, v).unwrap();
let out = server.conn.process_output(now()); let out = client.process(out.dgram(), now());
mem::drop(server.conn.process(out.dgram(), now()));
// send DATA frame before a header frame #[test] fn data_frame_on_push_stream() {
test_wrong_frame_on_push_stream(&[0x0, 0x2, 0x1, 0x2]);
}
// Client: receive unknown stream type // This function also tests getting stream id that does not fit into a single byte. #[test] fn client_received_unknown_stream() { let (mut client, mut server) = connect();
// create a stream with unknown type. let new_stream_id = server.conn.stream_create(StreamType::UniDi).unwrap();
_ = server
.conn
.stream_send(new_stream_id, &[0x41, 0x19, 0x4, 0x4, 0x6, 0x0, 0x8, 0x0])
.unwrap(); let out = server.conn.process_output(now()); let out = client.process(out.dgram(), now());
mem::drop(server.conn.process(out.dgram(), now()));
// Generate packet with the above bad h3 input let out = server.conn.process_output(now()); // Process bad input and close the connection.
mem::drop(client.process(out.dgram(), now()));
// Test reading of a slowly streamed frame. bytes are received one by one #[test] fn frame_reading() { let (mut client, mut server) = connect_only_transport();
// create a control stream. let control_stream = server.conn.stream_create(StreamType::UniDi).unwrap();
// send the stream type letmut sent = server.conn.stream_send(control_stream, &[0x0]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// start sending SETTINGS frame
sent = server.conn.stream_send(control_stream, &[0x4]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x4]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x6]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x0]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x8]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x0]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Now test PushPromise
sent = server.conn.stream_send(control_stream, &[0x5]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x5]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x4]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x61]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x62]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x63]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
sent = server.conn.stream_send(control_stream, &[0x64]);
assert_eq!(sent, Ok(1)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// PUSH_PROMISE on a control stream will cause an error
assert_closed(&client, &Error::HttpFrameUnexpected);
}
#[test] fn fetch_basic() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// send response - 200 Content-Length: 7 // with content: 'abcdefg'. // The content will be send in 2 DATA frames.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_1, true,
);
let http_events = client.events().collect::<Vec<_>>();
assert_eq!(http_events.len(), 2); for e in http_events { match e {
Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} => {
assert_eq!(stream_id, request_stream_id);
check_response_header_1(&headers);
assert!(!fin);
assert!(!interim);
}
Http3ClientEvent::DataReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id); letmut buf = [0_u8; 100]; let (amount, fin) = client.read_data(now(), stream_id, &mut buf).unwrap();
assert!(fin);
assert_eq!(amount, EXPECTED_RESPONSE_DATA_1.len());
assert_eq!(&buf[..amount], EXPECTED_RESPONSE_DATA_1);
}
_ => {}
}
}
// after this stream will be removed from hcoon. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
client.close(now(), 0, "");
}
/// Force both endpoints into an idle state. /// Do this by opening unidirectional streams at both endpoints and sending /// a partial unidirectional stream type (which the receiver has to buffer), /// then delivering packets out of order. /// This forces the receiver to create an acknowledgment, which will allow /// the peer to become idle. fn force_idle(client: &mut Http3Client, server: &mut TestServer) { // Send a partial unidirectional stream ID. // Note that this can't close the stream as that causes the receiver // to send `MAX_STREAMS`, which would prevent it from becoming idle. fn dgram(c: &mut Connection) -> Datagram { let stream = c.stream_create(StreamType::UniDi).unwrap();
_ = c.stream_send(stream, &[0xc0]).unwrap();
c.process_output(now()).dgram().unwrap()
}
let d1 = dgram(&mut client.conn); let d2 = dgram(&mut client.conn);
server.conn.process_input(d2, now());
server.conn.process_input(d1, now()); let d3 = dgram(&mut server.conn); let d4 = dgram(&mut server.conn);
client.process_input(d4, now());
client.process_input(d3, now()); let ack = client.process_output(now()).dgram();
server.conn.process_input(ack.unwrap(), now());
}
/// The client should keep a connection alive if it has unanswered requests. #[test] fn fetch_keep_alive() { let (mut client, mut server, _request_stream_id) = connect_and_send_request(true);
force_idle(&mut client, &mut server);
let idle_timeout = ConnectionParameters::default().get_idle_timeout();
assert_eq!(client.process_output(now()).callback(), idle_timeout / 2);
}
// Helper function: read response when a server sends HTTP_RESPONSE_2. fn read_response(
client: &mut Http3Client,
server: &mut Connection,
request_stream_id: StreamId,
) { let out = server.process_output(now());
client.process(out.dgram(), now());
// after this stream will be removed from client. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
client.close(now(), 0, "");
}
// Data sent with a request: const REQUEST_BODY: &[u8] = &[0x64, 0x65, 0x66]; // Corresponding data frame that server will receive. const EXPECTED_REQUEST_BODY_FRAME: &[u8] = &[0x0, 0x3, 0x64, 0x65, 0x66];
// Send a request with the request body. #[test] fn fetch_with_data() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Get DataWritable for the request stream so that we can write the request body. let data_writable = |e| matches!(e, Http3ClientEvent::DataWritable { .. });
assert!(client.events().any(data_writable)); let sent = client.send_data(request_stream_id, REQUEST_BODY).unwrap();
assert_eq!(sent, REQUEST_BODY.len());
client.stream_close_send(request_stream_id).unwrap();
let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// find the new request/response stream and send response on it. whilelet Some(e) = server.conn.next_event() { match e {
ConnectionEvent::NewStream { stream_id } => {
assert_eq!(stream_id, request_stream_id);
assert_eq!(stream_id.stream_type(), StreamType::BiDi);
}
ConnectionEvent::RecvStreamReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id);
// send a request with request body containing request_body. We expect to receive // expected_data_frame_header. fn fetch_with_data_length_xbytes(request_body: &[u8], expected_data_frame_header: &[u8]) { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Get DataWritable for the request stream so that we can write the request body. let data_writable = |e| matches!(e, Http3ClientEvent::DataWritable { .. });
assert!(client.events().any(data_writable)); let sent = client.send_data(request_stream_id, request_body);
assert_eq!(sent, Ok(request_body.len()));
// Close stream.
client.stream_close_send(request_stream_id).unwrap();
// We need to loop a bit until all data has been sent. letmut out = client.process_output(now()); for _i in0..20 {
out = server.conn.process(out.dgram(), now());
out = client.process(out.dgram(), now());
}
// check request body is received. // Then send a response. whilelet Some(e) = server.conn.next_event() { iflet ConnectionEvent::RecvStreamReadable { stream_id } = e { if stream_id == request_stream_id { // Read the DATA frame. letmut buf = vec![1_u8; RECV_BUFFER_SIZE]; let (amount, fin) = server.conn.stream_recv(stream_id, &mut buf).unwrap();
assert!(fin);
assert_eq!(
amount,
request_body.len() + expected_data_frame_header.len()
);
// Check the DATA frame header
assert_eq!(
&buf[..expected_data_frame_header.len()],
expected_data_frame_header
);
// send a request with 63 bytes. The DATA frame length field will still have 1 byte. #[test] fn fetch_with_data_length_63bytes() {
fetch_with_data_length_xbytes(&[0_u8; 63], &[0x0, 0x3f]);
}
// send a request with 64 bytes. The DATA frame length field will need 2 byte. #[test] fn fetch_with_data_length_64bytes() {
fetch_with_data_length_xbytes(&[0_u8; 64], &[0x0, 0x40, 0x40]);
}
// send a request with 16383 bytes. The DATA frame length field will still have 2 byte. #[test] fn fetch_with_data_length_16383bytes() {
fetch_with_data_length_xbytes(&[0_u8; 16383], &[0x0, 0x7f, 0xff]);
}
// send a request with 16384 bytes. The DATA frame length field will need 4 byte. #[test] fn fetch_with_data_length_16384bytes() {
fetch_with_data_length_xbytes(&[0_u8; 16384], &[0x0, 0x80, 0x0, 0x40, 0x0]);
}
// Send 2 data frames so that the second one cannot fit into the send_buf and it is only // partialy sent. We check that the sent data is correct. fn fetch_with_two_data_frames(
first_frame: &[u8],
expected_first_data_frame_header: &[u8],
expected_second_data_frame_header: &[u8],
expected_second_data_frame: &[u8],
) { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Get DataWritable for the request stream so that we can write the request body. let data_writable = |e| matches!(e, Http3ClientEvent::DataWritable { .. });
assert!(client.events().any(data_writable));
// Send the first frame. let sent = client.send_data(request_stream_id, first_frame);
assert_eq!(sent, Ok(first_frame.len()));
// The second frame cannot fit. let sent = client.send_data(request_stream_id, &vec![0_u8; SEND_BUFFER_SIZE]);
assert_eq!(sent, Ok(expected_second_data_frame.len()));
// Close stream.
client.stream_close_send(request_stream_id).unwrap();
letmut out = client.process_output(now()); // We need to loop a bit until all data has been sent. Once for every 1K // of data. for _i in0..SEND_BUFFER_SIZE / 1000 {
out = server.conn.process(out.dgram(), now());
out = client.process(out.dgram(), now());
}
// Check received frames and send a response. whilelet Some(e) = server.conn.next_event() { iflet ConnectionEvent::RecvStreamReadable { stream_id } = e { if stream_id == request_stream_id { // Read DATA frames. letmut buf = vec![1_u8; RECV_BUFFER_SIZE]; let (amount, fin) = server.conn.stream_recv(stream_id, &mut buf).unwrap();
assert!(fin);
assert_eq!(
amount,
expected_first_data_frame_header.len()
+ first_frame.len()
+ expected_second_data_frame_header.len()
+ expected_second_data_frame.len()
);
// Check the first DATA frame header let end = expected_first_data_frame_header.len();
assert_eq!(&buf[..end], expected_first_data_frame_header);
// Check the first frame data. let start = end; let end = end + first_frame.len();
assert_eq!(&buf[start..end], first_frame);
// Check the second DATA frame header let start2 = end; let end2 = end + expected_second_data_frame_header.len();
assert_eq!(&buf[start2..end2], expected_second_data_frame_header);
// Check the second frame data. let start3 = end2; let end3 = end2 + expected_second_data_frame.len();
assert_eq!(&buf[start3..end3], expected_second_data_frame);
// Send 2 frames. For the second one we can only send 63 bytes. // After the first frame there is exactly 63+2 bytes left in the send buffer. #[test] fn fetch_two_data_frame_second_63bytes() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 88);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x3f], &[yle='color: green'>0_u8; 63]);
}
// Send 2 frames. For the second one we can only send 63 bytes. // After the first frame there is exactly 63+3 bytes left in the send buffer, // but we can only send 63 bytes. #[test] fn fetch_two_data_frame_second_63bytes_place_for_66() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 89);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x3f], &[yle='color: green'>0_u8; 63]);
}
// Send 2 frames. For the second one we can only send 64 bytes. // After the first frame there is exactly 64+3 bytes left in the send buffer, // but we can only send 64 bytes. #[test] fn fetch_two_data_frame_second_64bytes_place_for_67() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 90);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x40, 0x40], &[an style='color: green'>0_u8; 64]);
}
// Send 2 frames. For the second one we can only send 16383 bytes. // After the first frame there is exactly 16383+3 bytes left in the send buffer. #[test] fn fetch_two_data_frame_second_16383bytes() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16409);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[an style='color: green'>0_u8; 16383]);
}
// Send 2 frames. For the second one we can only send 16383 bytes. // After the first frame there is exactly 16383+4 bytes left in the send buffer, but we can only // send 16383 bytes. #[test] fn fetch_two_data_frame_second_16383bytes_place_for_16387() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16410);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[an style='color: green'>0_u8; 16383]);
}
// Send 2 frames. For the second one we can only send 16383 bytes. // After the first frame there is exactly 16383+5 bytes left in the send buffer, but we can only // send 16383 bytes. #[test] fn fetch_two_data_frame_second_16383bytes_place_for_16388() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16411);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[an style='color: green'>0_u8; 16383]);
}
// Send 2 frames. For the second one we can send 16384 bytes. // After the first frame there is exactly 16384+5 bytes left in the send buffer, but we can send // 16384 bytes. #[test] fn fetch_two_data_frame_second_16384bytes_place_for_16389() { let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16412);
fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x80, 0x0, 0x40, 0x0], &[0_u8; 16384]);
}
// Test receiving STOP_SENDING with the HttpNoError error code. #[test] fn stop_sending_early_response() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Stop sending with early_response.
assert_eq!(
Ok(()),
server
.conn
.stream_stop_sending(request_stream_id, Error::HttpNoError.code())
);
// after this stream will be removed from client. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
client.close(now(), 0, "");
}
// Server sends stop sending and reset. #[test] fn stop_sending_other_error_with_reset() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Stop sending with RequestRejected.
assert_eq!(
Ok(()),
server
.conn
.stream_stop_sending(request_stream_id, Error::HttpRequestRejected.code())
); // also reset with RequestRejected.
assert_eq!(
Ok(()),
server
.conn
.stream_reset_send(request_stream_id, Error::HttpRequestRejected.code())
);
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// after this stream will be removed from client. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
client.close(now(), 0, "");
}
// Server sends stop sending with RequestRejected, but it does not send reset. #[test] fn stop_sending_other_error_wo_reset() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Stop sending with RequestRejected.
assert_eq!(
Ok(()),
server
.conn
.stream_stop_sending(request_stream_id, Error::HttpRequestRejected.code())
);
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
letmut stop_sending = false;
whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::StopSending { stream_id, error } => {
assert_eq!(stream_id, request_stream_id);
assert_eq!(error, Error::HttpRequestRejected.code());
stop_sending = true;
}
Http3ClientEvent::Reset { .. } => {
panic!("We should not get StopSending.");
}
Http3ClientEvent::HeaderReady { .. } | Http3ClientEvent::DataReadable { .. } => {
panic!("We should not get any headers or data");
}
_ => {}
}
}
assert!(stop_sending);
// after this we can still read from a stream. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_ok());
client.close(now(), 0, "");
}
// Server sends stop sending and reset. We have some events for that stream already // in client.events. The events will be removed. #[test] fn stop_sending_and_reset_other_error_with_events() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// send response - 200 Content-Length: 3 // with content: 'abc'.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_2, false,
); // At this moment we have some new events, i.e. a HeadersReady event
// Send a stop sending and reset.
assert_eq!(
Ok(()),
server
.conn
.stream_stop_sending(request_stream_id, Error::HttpRequestCancelled.code())
);
assert_eq!(
Ok(()),
server
.conn
.stream_reset_send(request_stream_id, Error::HttpRequestCancelled.code())
);
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
letmut reset = false;
whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::StopSending { stream_id, error } => {
assert_eq!(stream_id, request_stream_id);
assert_eq!(error, Error::HttpRequestCancelled.code());
}
Http3ClientEvent::Reset {
stream_id,
error,
local,
} => {
assert_eq!(stream_id, request_stream_id);
assert_eq!(error, Error::HttpRequestCancelled.code());
assert!(!local);
reset = true;
}
Http3ClientEvent::HeaderReady { .. } | Http3ClientEvent::DataReadable { .. } => {
panic!("We should not get any headers or data");
}
_ => {}
}
}
assert!(reset);
// after this stream will be removed from client. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
client.close(now(), 0, "");
}
// Server sends stop sending with code that is not HttpNoError. // We have some events for that stream already in the client.events. // The events will be removed. #[test] fn stop_sending_other_error_with_events() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// send response - 200 Content-Length: 3 // with content: 'abc'.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_2, false,
); // At this moment we have some new event, i.e. a HeadersReady event
// Send a stop sending.
assert_eq!(
Ok(()),
server
.conn
.stream_stop_sending(request_stream_id, Error::HttpRequestCancelled.code())
);
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// after this, we can sill read data from a sttream. letmut buf = [0_u8; 100]; let (amount, fin) = client
.read_data(now(), request_stream_id, &mut buf)
.unwrap();
assert!(!fin);
assert_eq!(amount, EXPECTED_RESPONSE_DATA_2_FRAME_1.len());
assert_eq!(&buf[..amount], EXPECTED_RESPONSE_DATA_2_FRAME_1);
client.close(now(), 0, "");
}
// Server sends a reset. We will close sending side as well. #[test] fn reset_wo_stop_sending() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(false);
// Send a reset.
assert_eq!(
Ok(()),
server
.conn
.stream_reset_send(request_stream_id, Error::HttpRequestCancelled.code())
);
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
letmut reset = false;
whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::StopSending { .. } => {
panic!("We should not get StopSending.");
}
Http3ClientEvent::Reset {
stream_id,
error,
local,
} => {
assert_eq!(stream_id, request_stream_id);
assert_eq!(error, Error::HttpRequestCancelled.code());
assert!(!local);
reset = true;
}
Http3ClientEvent::HeaderReady { .. } | Http3ClientEvent::DataReadable { .. } => {
panic!("We should not get any headers or data");
}
_ => {}
}
}
assert!(reset);
// after this stream will be removed from client. We will check this by trying to read // from the stream and that should fail. letmut buf = [0_u8; 100]; let res = client.read_data(now(), request_stream_id, &mut buf);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
// test goaway #[test] fn goaway() { let (mut client, mut server) = connect(); let request_stream_id_1 = make_request(&mut client, false, &[]);
assert_eq!(request_stream_id_1, 0); let request_stream_id_2 = make_request(&mut client, false, &[]);
assert_eq!(request_stream_id_2, 4); let request_stream_id_3 = make_request(&mut client, false, &[]);
assert_eq!(request_stream_id_3, 8);
let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
_ = server
.conn
.stream_send(server.control_stream_id.unwrap(), &[0x7, 0x1, 0x8])
.unwrap();
// find the new request/response stream and send frame v on it. whilelet Some(e) = server.conn.next_event() { iflet ConnectionEvent::RecvStreamReadable { stream_id } = e { letmut buf = [0_u8; 100];
_ = server.conn.stream_recv(stream_id, &mut buf).unwrap(); if (stream_id == request_stream_id_1) || (stream_id == request_stream_id_2) { // send response - 200 Content-Length: 7 // with content: 'abcdefg'. // The content will be send in 2 DATA frames.
_ = server.conn.stream_send(stream_id, HTTP_RESPONSE_1).unwrap();
server.conn.stream_close_send(stream_id).unwrap();
}
}
} let out = server.conn.process_output(now());
client.process(out.dgram(), now());
_ = server
.conn
.stream_send(server.control_stream_id.unwrap(), &[0x7, 0x1, 0x9])
.unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpId);
}
// Close stream before headers. #[test] fn stream_fin_wo_headers() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true); // send fin before sending any data.
server.conn.stream_close_send(request_stream_id).unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Recv HeaderReady wo headers with fin. let e = client.events().next().unwrap();
assert_eq!(
e,
Http3ClientEvent::Reset {
stream_id: request_stream_id,
error: Error::HttpGeneralProtocolStream.code(),
local: true,
}
);
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
// Close stream imemediately after headers. #[test] fn stream_fin_after_headers() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Recv HeaderReady with headers and fin. let e = client.events().next().unwrap(); iflet Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} = e
{
assert_eq!(stream_id, request_stream_id);
check_response_header_2(&headers);
assert!(fin);
assert!(!interim);
} else {
panic!("wrong event type");
}
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
// Send headers, read headers and than close stream. // We should get HeaderReady and a DataReadable #[test] fn stream_fin_after_headers_are_read_wo_data_frame() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true); // Send some good data wo fin
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_HEADER_ONLY_2, false,
);
// Recv headers wo fin whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} => {
assert_eq!(stream_id, request_stream_id);
check_response_header_2(&headers);
assert!(!fin);
assert!(!interim);
}
Http3ClientEvent::DataReadable { .. } => {
panic!("We should not receive a DataGeadable event!");
}
_ => {}
};
}
// ok NOW send fin
server.conn.stream_close_send(request_stream_id).unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Recv DataReadable wo data with fin whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::HeaderReady { .. } => {
panic!("We should not get another HeaderReady!");
}
Http3ClientEvent::DataReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id); letmut buf = [0_u8; 100]; let res = client.read_data(now(), stream_id, &mut buf); let (len, fin) = res.expect("should read");
assert_eq!(0, len);
assert!(fin);
}
_ => {}
};
}
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
// Send headers and an empty data frame, then close the stream. #[test] fn stream_fin_after_headers_and_a_empty_data_frame() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Send headers.
_ = server
.conn
.stream_send(request_stream_id, HTTP_RESPONSE_HEADER_ONLY_2)
.unwrap(); // Send an empty data frame.
_ = server
.conn
.stream_send(request_stream_id, &[0x00, 0x00])
.unwrap(); // ok NOW send fin
server.conn.stream_close_send(request_stream_id).unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), request_stream_id, &mut buf),
Err(Error::InvalidStreamId)
);
}
// Send headers and an empty data frame. Read headers and then close the stream. // We should get a HeaderReady without fin and a DataReadable wo data and with fin. #[test] fn stream_fin_after_headers_an_empty_data_frame_are_read() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true); // Send some good data wo fin // Send headers.
_ = server
.conn
.stream_send(request_stream_id, HTTP_RESPONSE_HEADER_ONLY_2)
.unwrap(); // Send an empty data frame.
_ = server
.conn
.stream_send(request_stream_id, &[0x00, 0x00])
.unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Recv headers wo fin whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} => {
assert_eq!(stream_id, request_stream_id);
check_response_header_2(&headers);
assert!(!fin);
assert!(!interim);
}
Http3ClientEvent::DataReadable { .. } => {
panic!("We should not receive a DataGeadable event!");
}
_ => {}
};
}
// ok NOW send fin
server.conn.stream_close_send(request_stream_id).unwrap();
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Recv no data, but do get fin whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::HeaderReady { .. } => {
panic!("We should not get another HeaderReady!");
}
Http3ClientEvent::DataReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id); letmut buf = [0_u8; 100]; let res = client.read_data(now(), stream_id, &mut buf); let (len, fin) = res.expect("should read");
assert_eq!(0, len);
assert!(fin);
}
_ => {}
};
}
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
#[test] fn stream_fin_after_a_data_frame() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true); // Send some good data wo fin
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_2, false,
);
// Recv some good data wo fin whilelet Some(e) = client.next_event() { match e {
Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} => {
assert_eq!(stream_id, request_stream_id);
check_response_header_2(&headers);
assert!(!fin);
assert!(!interim);
}
Http3ClientEvent::DataReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id); letmut buf = [0_u8; 100]; let res = client.read_data(now(), stream_id, &mut buf); let (len, fin) = res.expect("should have data");
assert_eq!(len, EXPECTED_RESPONSE_DATA_2_FRAME_1.len());
assert_eq!(&buf[..len], EXPECTED_RESPONSE_DATA_2_FRAME_1);
assert!(!fin);
}
_ => {}
};
}
// ok NOW send fin
server.conn.stream_close_send(request_stream_id).unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// fin wo data should generate DataReadable let e = client.events().next().unwrap(); iflet Http3ClientEvent::DataReadable { stream_id } = e {
assert_eq!(stream_id, request_stream_id); letmut buf = [0; 100]; let res = client.read_data(now(), stream_id, &mut buf); let (len, fin) = res.expect("should read");
assert_eq!(0, len);
assert!(fin);
} else {
panic!("wrong event type");
}
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
// Send a headers and a data frame with fin
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_2, true,
);
// Read first frame match client.events().nth(1).unwrap() {
Http3ClientEvent::DataReadable { stream_id } => {
assert_eq!(stream_id, request_stream_id); letmut buf = [0_u8; 100]; let (len, fin) = client.read_data(now(), stream_id, &mut buf).unwrap();
assert_eq!(len, EXPECTED_RESPONSE_DATA_2_FRAME_1.len());
assert_eq!(&buf[..len], EXPECTED_RESPONSE_DATA_2_FRAME_1);
assert!(fin);
}
x => {
panic!("event {x:?}");
}
} // Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
let headers = vec![
Header::new(":status", "200"),
Header::new("my-header", "my-header"),
Header::new("content-length", "3"),
]; let encoded_headers = server.encoder.borrow_mut().encode_header_block(
&mut server.conn,
&headers,
request_stream_id,
); let hframe = HFrame::Headers {
header_block: encoded_headers.to_vec(),
};
// Send the encoder instructions, but delay them so that the stream is blocked on decoding // headers. let encoder_inst_pkt = server.conn.process_output(now());
let sent_headers = vec![
Header::new(":status", "200"),
Header::new("my-header", "my-header"),
Header::new("content-length", "0"),
]; let encoded_headers = server.encoder.borrow_mut().encode_header_block(
&mut server.conn,
&sent_headers,
request_stream_id,
); let hframe = HFrame::Headers {
header_block: encoded_headers.to_vec(),
};
// Send the encoder instructions, but delay them so that the stream is blocked on decoding // headers. let encoder_inst_pkt = server.conn.process_output(now());
letmut d = Encoder::default();
hframe.encode(&mut d);
let header_ready_event = |e| matches!(e, Http3ClientEvent::HeaderReady { .. });
assert!(!hconn.events().any(header_ready_event));
// Let client receive the encoder instructions. let _out = hconn.process(encoder_inst_pkt.dgram(), now());
letmut recv_header = false; // Now the stream is unblocked. After headers we will receive a fin. whilelet Some(e) = hconn.next_event() { iflet Http3ClientEvent::HeaderReady {
stream_id,
headers,
interim,
fin,
} = e
{
assert_eq!(stream_id, request_stream_id);
assert_eq!(headers.as_ref(), sent_headers);
assert!(fin);
assert!(!interim);
recv_header = true;
} else {
panic!("event {e:?}");
}
}
assert!(recv_header);
}
fn exchange_token(client: &mut Http3Client, server: &'color:red'>mut Connection) -> ResumptionToken {
server.send_ticket(now(), &[]).expect("can send ticket"); let out = server.process_output(now());
assert!(out.as_dgram_ref().is_some());
client.process_input(out.dgram().unwrap(), now()); // We do not have a token so we need to wait for a resumption token timer to trigger.
client.process_output(now() + Duration::from_millis(250));
assert_eq!(client.state(), Http3State::Connected);
client
.events()
.find_map(|e| { iflet Http3ClientEvent::ResumptionToken(token) = e {
Some(token)
} else {
None
}
})
.unwrap()
}
fn start_with_0rtt() -> (Http3Client, TestServer) { let (mut client, mut server) = connect(); let token = exchange_token(&mut client, &mut server.conn);
assert_eq!(client.state(), Http3State::ZeroRtt);
assert_eq!(*server.conn.state(), State::Init); let out = server.conn.process(out.dgram(), now());
// Check that control and qpack streams are received and a // SETTINGS frame has been received. // Also qpack encoder stream will send "change capacity" instruction because it has // the peer settings already.
server.check_control_qpack_request_streams_resumption(
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
EXPECTED_REQUEST_HEADER_FRAME, false,
);
assert_eq!(*server.conn.state(), State::Handshaking); let out = client.process(out.dgram(), now());
assert_eq!(client.state(), Http3State::Connected);
let request_stream_id =
make_request(&mut client, true, &[Header::new("myheaders", "myvalue")]);
assert_eq!(request_stream_id, 0);
let out = client.process_output(now());
assert_eq!(client.state(), Http3State::ZeroRtt);
assert_eq!(*server.conn.state(), State::Init); let out = server.conn.process(out.dgram(), now());
// Check that control and qpack streams are received and a // SETTINGS frame has been received. // Also qpack encoder stream will send "change capacity" instruction because it has // the peer settings already.
server.check_control_qpack_request_streams_resumption(
ENCODER_STREAM_DATA_WITH_CAP_INST_AND_ENCODING_INST,
EXPECTED_REQUEST_HEADER_FRAME_VERSION2, true,
);
assert_eq!(*server.conn.state(), State::Handshaking); let out = client.process(out.dgram(), now());
assert_eq!(client.state(), Http3State::Connected); let out = server.conn.process(out.dgram(), now());
assert!(server.conn.state().connected()); let out = client.process(out.dgram(), now());
assert!(out.dgram().is_none());
// After the server has been connected, send a response. let res = server.conn.stream_send(request_stream_id, HTTP_RESPONSE_2);
assert_eq!(res, Ok(HTTP_RESPONSE_2.len()));
server.conn.stream_close_send(request_stream_id).unwrap();
#[test] fn zero_rtt_send_reject() { let (mut client, mut server) = connect(); let token = exchange_token(&mut client, &mut server.conn);
letmut client = default_http3_client(); letmut server = Connection::new_server(
test_fixture::DEFAULT_KEYS,
test_fixture::DEFAULT_ALPN_H3,
Rc::new(RefCell::new(CountingConnectionIdGenerator::default())),
ConnectionParameters::default(),
)
.unwrap(); // Using a freshly initialized anti-replay context // should result in the server rejecting 0-RTT. let ar = AntiReplay::new(now(), test_fixture::ANTI_REPLAY_WINDOW, 1, 3)
.expect("setup anti-replay");
server
.server_enable_0rtt(&ar, AllowZeroRtt {})
.expect("enable 0-RTT");
// Send ClientHello. let client_hs = client.process_output(now());
assert!(client_hs.as_dgram_ref().is_some());
// Create a request let request_stream_id = make_request(&mut client, false, &[]);
assert_eq!(request_stream_id, 0);
let client_0rtt = client.process_output(now());
assert!(client_0rtt.as_dgram_ref().is_some());
let server_hs = server.process(client_hs.dgram(), now());
assert!(server_hs.as_dgram_ref().is_some()); // Should produce ServerHello etc... let server_ignored = server.process(client_0rtt.dgram(), now());
assert!(server_ignored.as_dgram_ref().is_none());
// The server shouldn't receive that 0-RTT data. let recvd_stream_evt = |e| matches!(e, ConnectionEvent::NewStream { .. });
assert!(!server.events().any(recvd_stream_evt));
// Client should get a rejection. let client_out = client.process(server_hs.dgram(), now());
assert!(client_out.as_dgram_ref().is_some()); let recvd_0rtt_reject = |e| e == Http3ClientEvent::ZeroRttRejected;
assert!(client.events().any(recvd_0rtt_reject));
// ...and the client stream should be gone. let res = client.stream_close_send(request_stream_id);
assert!(res.is_err());
assert_eq!(res.unwrap_err(), Error::InvalidStreamId);
// Client will send Setting frame and open new qpack streams.
mem::drop(server.process(client_out.dgram(), now()));
TestServer::new_with_conn(server).check_client_control_qpack_streams_no_resumption();
// Check that we can send a request and that the stream_id starts again from 0.
assert_eq!(make_request(&mut client, false, &[]), 0);
}
// Connect to a server, get token and reconnect using 0-rtt. Seerver sends new Settings. fn zero_rtt_change_settings(
original_settings: &[HSetting],
resumption_settings: &[HSetting],
expected_client_state: &Http3State,
expected_encoder_stream_data: &[u8],
) { letmut client = default_http3_client(); letmut server = TestServer::new_with_settings(original_settings); // Connect and get a token
connect_with(&mut client, &mut server); let token = exchange_token(&mut client, &mut server.conn);
letmut client = default_http3_client(); letmut server = TestServer::new_with_settings(resumption_settings);
assert_eq!(client.state(), Http3State::Initializing);
client
.enable_resumption(now(), &token)
.expect("Set resumption token.");
assert_eq!(client.state(), Http3State::ZeroRtt); let out = client.process_output(now());
assert_eq!(client.state(), Http3State::ZeroRtt);
assert_eq!(*server.conn.state(), State::Init); let out = server.conn.process(out.dgram(), now());
// Check that control and qpack streams and a SETTINGS frame are received. // Also qpack encoder stream will send "change capacity" instruction because it has // the peer settings already.
server.check_control_qpack_request_streams_resumption(
expected_encoder_stream_data,
EXPECTED_REQUEST_HEADER_FRAME, false,
);
assert_eq!(*server.conn.state(), State::Handshaking); let out = client.process(out.dgram(), now());
assert_eq!(client.state(), Http3State::Connected);
#[test] fn zero_rtt_new_server_setting_are_the_same() { // Send a new server settings that are the same as the old one.
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Connected,
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_new_server_setting_omit_max_table() { // Send a new server settings without MaxTableCapacity
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Closing(CloseReason::Application(265)),
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_new_server_setting_omit_blocked_streams() { // Send a new server settings without BlockedStreams
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Closing(CloseReason::Application(265)),
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_new_server_setting_omit_header_list_size() { // Send a new server settings without MaxHeaderListSize
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
],
&Http3State::Connected,
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_new_server_setting_max_header_size_bigger() { // Send a new server settings with MaxHeaderListSize=20000
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 20000),
],
&Http3State::Connected,
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_new_server_setting_max_headers_size_smaller() { // Send the new server settings with MaxHeaderListSize=5000
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 5000),
],
&Http3State::Closing(CloseReason::Application(265)),
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_max_table_size_first_omitted() { // send server original settings without MaxTableCapacity // send new server setting with MaxTableCapacity
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Connected,
ENCODER_STREAM_DATA,
);
}
#[test] fn zero_rtt_blocked_streams_first_omitted() { // Send server original settings without BlockedStreams // Send the new server settings with BlockedStreams
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Connected,
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn zero_rtt_max_header_size_first_omitted() { // Send server settings without MaxHeaderListSize // Send new settings with MaxHeaderListSize.
zero_rtt_change_settings(
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 10000),
],
&[
HSetting::new(HSettingType::MaxTableCapacity, 100),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
],
&Http3State::Closing(CloseReason::Application(265)),
ENCODER_STREAM_DATA_WITH_CAP_INSTRUCTION,
);
}
#[test] fn trailers_with_fin_after_headers() { // Make a new connection. let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
let events: Vec<Http3ClientEvent> = client.events().collect();
// We already had HeaderReady let header_ready: fn(&Http3ClientEvent) -> _ =
|e| matches!(*e, Http3ClientEvent::HeaderReady { .. });
assert!(!events.iter().any(header_ready));
// Check that we have a DataReady event. Reading from the stream will return fin=true. let data_readable: fn(&Http3ClientEvent) -> _ =
|e| matches!(*e, Http3ClientEvent::DataReadable { .. });
assert!(events.iter().any(data_readable)); letmut buf = [0_u8; 100]; let (len, fin) = client
.read_data(now(), request_stream_id, &mut buf)
.unwrap();
assert_eq!(0, len);
assert!(fin);
}
#[test] fn trailers_with_later_fin_after_headers() { // Make a new connection. let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Check that we do not have a DataReady event. let data_readable = |e| matches!(e, Http3ClientEvent::DataReadable { .. });
assert!(!client.events().any(data_readable));
let out = server.conn.process_output(now());
client.process(out.dgram(), now());
let events: Vec<Http3ClientEvent> = client.events().collect();
// We already had HeaderReady let header_ready: fn(&Http3ClientEvent) -> _ =
|e| matches!(*e, Http3ClientEvent::HeaderReady { .. });
assert!(!events.iter().any(header_ready));
// Check that we have a DataReady event. Reading from the stream will return fin=true. let data_readable_fn: fn(&Http3ClientEvent) -> _ =
|e| matches!(*e, Http3ClientEvent::DataReadable { .. });
assert!(events.iter().any(data_readable_fn)); letmut buf = [0_u8; 100]; let (len, fin) = client
.read_data(now(), request_stream_id, &mut buf)
.unwrap();
assert_eq!(0, len);
assert!(fin);
}
#[test] fn data_after_trailers_after_headers() { // Make a new connection. let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Check that we do not have a DataReady event. let data_readable = |e| matches!(e, Http3ClientEvent::DataReadable { .. });
assert!(!client.events().any(data_readable));
// Send Data frame.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
[0x0, 0x3, 0x61, 0x62, 0x63], // a data frame false,
);
#[test] fn no_data_ready_events_after_fin() { // Connect exchange headers and send a request. Also check if the correct header frame has // been sent. let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// send response - 200 Content-Length: 7 // with content: 'abcdefg'. // The content will be send in 2 DATA frames.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_1, true,
);
// send response - 200 Content-Length: 7 // with content: 'abcdefg'. // The content will be send in 2 DATA frames.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_1, true,
);
// Now read only until the end of the first frame. The firs framee has 3 bytes. letmut buf2 = [0_u8; 2];
assert_eq!(
(2, false),
client
.read_data(now(), request_stream_id, &mut buf2)
.unwrap()
);
assert!(!client.events().any(data_readable_event));
// Read a half of the second frame.
assert_eq!(
(2, false),
client
.read_data(now(), request_stream_id, &mut buf2)
.unwrap()
);
assert!(!client.events().any(data_readable_event));
// Read the rest. // Read a half of the second frame.
assert_eq!(
(2, true),
client
.read_data(now(), request_stream_id, &mut buf2)
.unwrap()
);
assert!(!client.events().any(data_readable_event));
}
// send response - 200 Content-Length: 7 // with content: 'abcdefg'. // The content will be send in 2 DATA frames.
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_1, false,
); // Send a zero-length frame at the end of the stream.
_ = server.conn.stream_send(request_stream_id, &[0, 0]).unwrap();
server.conn.stream_close_send(request_stream_id).unwrap(); let dgram = server.conn.process_output(now()).dgram();
client.process_input(dgram.unwrap(), now());
let data_readable_event = |e: &_| matches!(e, Http3ClientEvent::DataReadable { stream_id } if *stream_id == request_stream_id);
assert_eq!(client.events().filter(data_readable_event).count(), 1);
server.create_control_stream(); // Send the server's control stream data. let out = server.conn.process_output(now());
client.process(out.dgram(), now());
server.create_qpack_streams(); let qpack_pkt1 = server.conn.process_output(now()); // delay delivery of this packet.
let request_stream_id = make_request(&mut client, true, &[]); let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
}
/// We can't keep the connection alive on the basis of a push promise, /// nor do we want to if the push promise is not interesting to the client. /// We do the next best thing, which is keep any push stream alive if the /// client reads from it. #[test] fn push_keep_alive() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true); let idle_timeout = ConnectionParameters::default().get_idle_timeout();
// Promise a push and deliver, but don't close the stream.
send_push_promise(&mut server.conn, request_stream_id, 0);
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_2, true,
);
read_response_and_push_events(
&mut client,
&[PushPromiseInfo {
push_id: 0,
ref_stream_id: request_stream_id,
}],
&[], // No push streams yet.
request_stream_id,
);
// The client will become idle here.
force_idle(&mut client, &mut server);
assert_eq!(client.process_output(now()).callback(), idle_timeout);
// Reading push data will stop the client from being idle.
_ = send_push_data(&mut server.conn, 0, false); let out = server.conn.process_output(now());
client.process_input(out.dgram().unwrap(), now());
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
assert_eq!(client.cancel_push(1), Err(Error::InvalidStreamId));
}
#[test] fn push_after_headers() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Test receiving pushes out of order. // Push_id 5 is received first, therefore Push_id 3 will be in the PushState:Init state. // Start push_id 3 by receiving a push_promise and then a push stream with the push_id 3. #[test] fn push_out_of_order_1() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Test receiving pushes out of order. // Push_id 5 is received first, therefore Push_id 3 will be in the PushState:Init state. // Start push_id 3 by receiving a push stream with push_id 3 and then a push_promise. #[test] fn push_out_of_order_2() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Test receiving pushes out of order. // Push_id 5 is received first and read so that it is removed from the list, // therefore Push_id 3 will be in the PushState:Init state. // Start push_id 3 by receiving a push stream with the push_id 3 and then a push_promise. #[test] fn push_out_of_order_3() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// The next test is for receiving a second PushPromise when Push is in the PushPromise state. #[test] fn multiple_push_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// The next test is for receiving a second PushPromise when Push is in the Active state. #[test] fn multiple_push_promise_active() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// The next test is for receiving a second PushPromise when the push is already closed. // PushPromise will be ignored for the push streams that are consumed. #[test] fn multiple_push_promise_closed() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
send_push_promise_and_exchange_packets(&mut client, &le='color:red'>mut server, request_stream_id, 5); // Start a push stream with push_id 5.
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 5, true);
// Check that we do not have a Http3ClientEvent::PushPromise. let push_event = |e| matches!(e, Http3ClientEvent::PushPromise { .. });
assert!(!client.events().any(push_event));
}
// Test that max_push_id is enforced when a push promise frame is received. #[test] fn exceed_max_push_id_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Send a push promise. max_push_id is set to 5, to trigger an error we send push_id=6.
send_push_promise_and_exchange_packets(&mut client, &le='color:red'>mut server, request_stream_id, 6);
assert_closed(&client, &Error::HttpId);
}
// Test that max_push_id is enforced when a push stream is received. #[test] fn exceed_max_push_id_push_stream() { // Connect and send a request let (mut client, mut server) = connect();
// Send a push stream. max_push_id is set to 5, to trigger an error we send push_id=6.
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 6, true);
assert_closed(&client, &Error::HttpId);
}
// Test that max_push_id is enforced when a cancel push frame is received. #[test] fn exceed_max_push_id_cancel_push() { // Connect and send a request let (mut client, mut server, _request_stream_id) = connect_and_send_request(true);
// Test that max_push_id is enforced when an app calls cancel_push. #[test] fn exceed_max_push_id_cancel_api() { // Connect and send a request let (mut client, _, _) = connect_and_send_request(true);
let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// Check max_push_id frame has been received let control_stream_readable =
|e| matches!(e, ConnectionEvent::RecvStreamReadable{stream_id: x} if x == 2);
assert!(server.conn.events().any(control_stream_readable)); letmut buf = [0_u8; 100]; let (amount, fin) = server.conn.stream_recv(StreamId::new(2), &mut buf).unwrap();
assert!(!fin);
// Test that 2 push streams with the same push_id are caught. #[test] fn duplicate_push_stream() { // Connect and send a request let (mut client, mut server, _request_stream_id) = connect_and_send_request(true);
// Start a push stream with push_id 0.
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, true);
// Send it again
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, true);
assert_closed(&client, &Error::HttpId);
}
// Test that 2 push streams with the same push_id are caught. #[test] fn duplicate_push_stream_active() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
send_push_promise(&mut server.conn, request_stream_id, 0);
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, true); // Now the push_stream is in the PushState::Active state
// Test CANCEL_PUSH frame: after cancel push any new PUSH_PROMISE or push stream will be // ignored. #[test] fn cancel_push_ignore_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
send_push_promise(&mut server.conn, request_stream_id, 0); // Start a push stream with push_id 0. let push_stream_id =
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, false);
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Check that the push has been canceled by the client.
assert_stop_sending_event(
&mut server,
push_stream_id,
Error::HttpRequestCancelled.code(),
);
// Test CANCEL_PUSH frame: after cancel push any already received PUSH_PROMISE or push stream // events will be removed. #[test] fn cancel_push_removes_push_events() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Check that the push has been canceled by the client.
assert_stop_sending_event(
&mut server,
push_stream_id,
Error::HttpRequestCancelled.code(),
);
// Test CANCEL_PUSH frame: after cancel push any already received push stream will be canceled. #[test] fn cancel_push_frame_after_push_stream() { // Connect and send a request let (mut client, mut server, _) = connect_and_send_request(true);
// Start a push stream with push_id 0. let push_stream_id =
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, false);
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Check that the push has been canceled by the client.
assert_stop_sending_event(
&mut server,
push_stream_id,
Error::HttpRequestCancelled.code(),
);
// Test a push stream reset after a new PUSH_PROMISE or/and push stream. The events will be // ignored. #[test] fn cancel_push_stream_after_push_promise_and_push_stream() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
send_push_promise(&mut server.conn, request_stream_id, 0); // Start a push stream with push_id 0. let push_stream_id =
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, false);
server
.conn
.stream_reset_send(push_stream_id, Error::HttpRequestCancelled.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Test that a PUSH_PROMISE will be ignored after a push stream reset. #[test] fn cancel_push_stream_before_push_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Start a push stream with push_id 0. let push_stream_id =
send_push_data_and_exchange_packets(&mut client, &'color:red'>mut server, 0, false);
server
.conn
.stream_reset_send(push_stream_id, Error::HttpRequestCancelled.code())
.unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Test that push_promise events will be removed after application calls cancel_push. #[test] fn app_cancel_push_after_push_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Test that push_promise and push data events will be removed after application calls // cancel_push. #[test] fn app_cancel_push_after_push_promise_and_push_stream() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
assert!(client.cancel_push(0).is_ok()); let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Check that the push has been canceled by the client.
assert_stop_sending_event(
&mut server,
push_stream_id,
Error::HttpRequestCancelled.code(),
);
// Test that push_promise events will be ignored after application calls cancel_push. #[test] fn app_cancel_push_before_push_promise() { // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Assert that we do not have any push event.
assert!(!check_push_events(&mut client));
// Check that the push has been closed, e.g. calling cancel_push should return // InvalidStreamId.
assert_eq!(client.cancel_push(0), Err(Error::InvalidStreamId));
// Check that the push has been canceled by the client.
assert_stop_sending_event(
&mut server,
push_stream_id,
Error::HttpRequestCancelled.code(),
);
let encoded_headers =
server
.encoder
.borrow_mut()
.encode_header_block(&mut server.conn, &headers, stream_id); let push_promise_frame = HFrame::PushPromise {
push_id,
header_block: encoded_headers.to_vec(),
};
// Send the encoder instructions, but delay them so that the stream is blocked on decoding // headers. let encoder_inst_pkt = server.conn.process_output(now()).dgram();
assert!(encoder_inst_pkt.is_some());
let encoder_inst_pkt =
send_push_promise_using_encoder(&mut client, &mut server, request_stream_id, 0);
// PushPromise is blocked wathing for encoder instructions.
assert!(!check_push_events(&mut client));
// Let client receive the encoder instructions. let _out = client.process(encoder_inst_pkt, now());
// PushPromise is blocked wathing for encoder instructions.
assert!(check_push_events(&mut client));
}
// If PushPromise is blocked, stream data can still be received. #[test] fn push_promise_blocked_but_stream_is_not_blocked() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
let encoder_inst_pkt =
send_push_promise_using_encoder(&mut client, &mut server, request_stream_id, 0);
// PushPromise is blocked wathing for encoder instructions.
assert!(!check_push_events(&mut client));
// Stream data can be still read
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_DATA_FRAME_1_ONLY_1, false,
);
assert!(check_data_readable(&mut client));
// Let client receive the encoder instructions. let _out = client.process(encoder_inst_pkt, now());
// PushPromise is blocked wathing for encoder instructions.
assert!(check_push_events(&mut client));
// Stream data can be still read
server_send_response_and_exchange_packet(
&mut client,
&mut server,
request_stream_id,
HTTP_RESPONSE_DATA_FRAME_2_ONLY_1, false,
);
assert!(check_data_readable(&mut client));
}
// The response Headers are not block if they do not refer the dynamic table. #[test] fn push_promise_does_not_block_headers() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Let client receive the encoder instructions. let _out = client.process(encoder_inst_pkt, now());
// PushPromise is blocked wathing for encoder instructions.
assert!(check_push_events(&mut client));
}
// The response Headers are blocked if they refer a dynamic table entry. #[test] fn push_promise_block_headers() { let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Insert an elemet into a dynamic table. // insert "content-length: 1234
server
.encoder
.borrow_mut()
.send_and_insert(&mut server.conn, b"content-length", b"1234")
.unwrap(); let encoder_inst_pkt1 = server.conn.process_output(now()).dgram(); let _out = client.process(encoder_inst_pkt1, now());
// Send a PushPromise that is blocked until encoder_inst_pkt2 is process by the client. let encoder_inst_pkt2 =
send_push_promise_using_encoder(&mut client, &mut server, request_stream_id, 0);
// PushPromise is blocked wathing for encoder instructions.
assert!(!check_push_events(&mut client));
// The response headers are blocked.
assert!(!check_header_ready(&mut client));
// Let client receive the encoder instructions. let _out = client.process(encoder_inst_pkt2, now());
// The response headers are blocked.
assert!(check_header_ready_and_push_promise(&mut client));
}
// In this test there are 2 push promises that are blocked and the response header is // blocked as well. After a packet is received only the first push promises is unblocked. #[test] fn two_push_promises_and_header_block() { letmut client = default_http3_client_param(200); letmut server = TestServer::new_with_settings(&[
HSetting::new(HSettingType::MaxTableCapacity, 200),
HSetting::new(HSettingType::BlockedStreams, 100),
HSetting::new(HSettingType::MaxHeaderListSize, 10000),
]);
connect_only_transport_with(&mut client, &mut server);
server.create_control_stream();
server.create_qpack_streams();
setup_server_side_encoder_param(&mut client, &mut server, 200);
let request_stream_id = make_request_and_exchange_pkts(&mut client, &mut server, true);
// Send a PushPromise that is blocked until encoder_inst_pkt2 is process by the client. let encoder_inst_pkt1 = send_push_promise_using_encoder_with_custom_headers(
&mut client,
&mut server,
request_stream_id, 0,
Header::new("myn1", "myv1"),
);
// PushPromise is blocked wathing for encoder instructions.
assert!(!check_push_events(&mut client));
// The PushPromise blocked on header decoding will be canceled if the stream is closed. #[test] fn blocked_push_promises_canceled() { const STREAM_CANCELED_ID_0: &[u8] = &[0x40];
let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
let header_ready_event = |e| matches!(e, Http3ClientEvent::HeaderReady { .. });
assert!(client.events().any(header_ready_event)); // After this the recv_stream is in ClosePending state
let headers = vec![
Header::new(":status", "200"),
Header::new("content-length", "3"),
]; let encoded_headers = server.encoder.borrow_mut().encode_header_block(
&mut server.conn,
&headers,
request_stream_id,
); let hframe = HFrame::Headers {
header_block: encoded_headers.to_vec(),
};
// Send the encoder instructions. let out = server.conn.process_output(now());
mem::drop(client.process(out.dgram(), now()));
// Send PushPromise that will be blocked waiting for decoder instructions.
mem::drop(
send_push_promise_using_encoder(&mut client, &mut server, request_stream_id, 0)
.unwrap(),
);
// Headers are blocked waiting for the encoder instructions. let header_ready_event = |e| matches!(e, Http3ClientEvent::HeaderReady { .. });
assert!(!client.events().any(header_ready_event));
// Make another request. let request2 = make_request_and_exchange_pkts(&mut client, & style='color:red'>mut server, true); // Send response headers.
server_send_response_and_exchange_packet(&mut client, &tyle='color:red'>mut server, request2, &d, true);
// Headers on the second request are blocked as well are blocked // waiting for the encoder instructions.
assert!(!client.events().any(header_ready_event));
// Now make the encoder instructions available.
mem::drop(client.process(encoder_insts.dgram(), now()));
// Header blocks for both streams should be ready. letmut count_responses = 0; whilelet Some(e) = client.next_event() { iflet Http3ClientEvent::HeaderReady { stream_id, .. } = e {
assert!((stream_id == request_stream_id) || (stream_id == request2));
count_responses += 1;
}
}
assert_eq!(count_responses, 2);
}
#[test] fn reserved_frames() { for f in H3_RESERVED_FRAME_TYPES { letmut enc = Encoder::default();
enc.encode_varint(*f);
test_wrong_frame_on_control_stream(enc.as_ref());
test_wrong_frame_on_push_stream(enc.as_ref());
test_wrong_frame_on_request_stream(enc.as_ref());
}
}
#[test] fn send_reserved_settings() { for s in H3_RESERVED_SETTINGS { let (mut client, mut server) = connect_only_transport(); let control_stream = server.conn.stream_create(StreamType::UniDi).unwrap(); // Send the control stream type(0x0).
_ = server
.conn
.stream_send(control_stream, CONTROL_STREAM_TYPE)
.unwrap(); // Create a settings frame of length 2. letmut enc = Encoder::default();
enc.encode_varint(H3_FRAME_TYPE_SETTINGS);
enc.encode_varint(2_u64); // The settings frame contains a reserved settings type and some value (0x1).
enc.encode_varint(*s);
enc.encode_varint(1_u64); let sent = server.conn.stream_send(control_stream, enc.as_ref());
assert_eq!(sent, Ok(4)); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpSettings);
}
}
// Stream has been reset because of the malformed headers. let e = client.events().next().unwrap();
assert_eq!(
e,
Http3ClientEvent::Reset {
stream_id: request_stream_id,
error: Error::InvalidHeader.code(),
local: true,
}
);
let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// Check that server has received a reset. let stop_sending_event = |e| {
matches!(e, ConnectionEvent::SendStreamStopSending {
stream_id,
app_error
} if stream_id == request_stream_id && app_error == Error::InvalidHeader.code())
};
assert!(server.conn.events().any(stop_sending_event));
// Stream should now be closed and gone letmut buf = [0_u8; 100];
assert_eq!(
client.read_data(now(), StreamId::new(0), &mut buf),
Err(Error::InvalidStreamId)
);
}
// Client: receive a push stream #[test] fn push_single_with_1xx() { const FIRST_PUSH_ID: u64 = 0; // Connect and send a request let (mut client, mut server, request_stream_id) = connect_and_send_request(true);
// Send a push promise.
send_push_promise(&mut server.conn, request_stream_id, FIRST_PUSH_ID); // Create a push stream let push_stream_id = server.conn.stream_create(StreamType::UniDi).unwrap();
letmut d = Encoder::default(); let headers1xx: &[Header] = &[Header::new(":status", "100")];
server.encode_headers(push_stream_id, headers1xx, &mut d);
// Stream has been reset because of thei malformed headers. let push_reset_event = |e| {
matches!(e, Http3ClientEvent::PushReset {
push_id,
error,
} if push_id == FIRST_PUSH_ID && error == Error::InvalidHeader.code())
};
assert!(client.events().any(push_reset_event));
let out = client.process_output(now());
mem::drop(server.conn.process(out.dgram(), now()));
// Check that server has received a reset. let stop_sending_event = |e| {
matches!(e, ConnectionEvent::SendStreamStopSending {
stream_id,
app_error
} if stream_id == push_stream_id && app_error == Error::InvalidHeader.code())
};
assert!(server.conn.events().any(stop_sending_event));
}
/// Client fails to create a control stream, since server does not allow it. #[test] fn client_control_stream_create_failed() { letmut client = default_http3_client(); letmut server = TestServer::new_with_conn(new_server(
DEFAULT_ALPN_H3,
ConnectionParameters::default().max_streams(StreamType::UniDi, 0),
));
handshake_client_error(&mut client, &mut server, &Error::StreamLimitError);
}
/// 2 streams isn't enough for control and QPACK streams. #[test] fn client_qpack_stream_create_failed() { letmut client = default_http3_client(); letmut server = TestServer::new_with_conn(new_server(
DEFAULT_ALPN_H3,
ConnectionParameters::default().max_streams(StreamType::UniDi, 2),
));
handshake_client_error(&mut client, &mut server, &Error::StreamLimitError);
}
// Stream has been reset because of the malformed headers. let e = client.events().next().unwrap();
assert_eq!(
e,
Http3ClientEvent::Reset {
stream_id: request_stream_id,
error: Error::InvalidHeader.code(),
local: true,
}
);
}
// Stream has been reset because of the malformed headers. let e = client.events().next().unwrap();
assert_eq!(
e,
Http3ClientEvent::HeaderReady {
stream_id: request_stream_id,
headers: vec![
Header::new(":status", "200"),
Header::new("content-type", "text/plain")
],
interim: false,
fin: false,
}
);
}
// Test that decoder stream type is always sent before any other instruction also // in case when 0RTT is used. // A client will send a request that uses the dynamic table. This will trigger a header-ack // from a server. We will use stats to check that a header-ack has been received. #[test] fn zerortt_request_use_dynamic_table() { letmut server = Http3Server::new(
now(),
DEFAULT_KEYS,
DEFAULT_ALPN_H3,
anti_replay(),
Rc::new(RefCell::new(CountingConnectionIdGenerator::default())),
Http3Parameters::default()
.max_table_size_encoder(MAX_TABLE_SIZE)
.max_table_size_decoder(MAX_TABLE_SIZE)
.max_blocked_streams(MAX_BLOCKED_STREAMS),
None,
)
.unwrap();
let token = get_resumption_token(&mut server); // Make a new connection. letmut client = default_http3_client_param(MAX_TABLE_SIZE);
assert_eq!(client.state(), Http3State::Initializing);
client
.enable_resumption(now(), &token)
.expect("Set resumption token.");
assert_eq!(client.state(), Http3State::ZeroRtt); let zerortt_event = |e| matches!(e, Http3ClientEvent::StateChange(Http3State::ZeroRtt));
assert!(client.events().any(zerortt_event));
// Make a request that uses the dynamic table.
_ = make_request(&mut client, true, &[Header::new("myheaders", "myvalue")]); // Assert that the request has used dynamic table. That will trigger a header_ack.
assert_eq!(client.qpack_encoder_stats().dynamic_table_references, 1);
// Exchange packets until header-ack is received. // These many packet exchange is needed, to get a header-ack. // TODO this may be optimize at Http3Server. let out = client.process_output(now()); let out = server.process(out.dgram(), now()); let out = client.process(out.dgram(), now()); let out = server.process(out.dgram(), now()); let out = client.process(out.dgram(), now()); let out = server.process(out.dgram(), now()); let out = client.process(out.dgram(), now()); let out = server.process(out.dgram(), now());
mem::drop(client.process(out.dgram(), now()));
// The header ack for the first request has been received.
assert_eq!(client.qpack_encoder_stats().header_acks_recv, 1);
}
// Client: receive a push stream #[test] fn incomple_push_stream() { let (mut client, mut server) = connect();
// Create a push stream let push_stream_id = server.conn.stream_create(StreamType::UniDi).unwrap();
_ = server
.conn
.stream_send(push_stream_id, PUSH_STREAM_TYPE)
.unwrap();
_ = server.conn.stream_send(push_stream_id, &[0]).unwrap();
server.conn.stream_close_send(push_stream_id).unwrap(); let out = server.conn.process_output(now());
client.process(out.dgram(), now());
assert_closed(&client, &Error::HttpGeneralProtocol);
}
#[test] fn priority_update_during_full_buffer() { // set a lower MAX_DATA on the server side to restrict the data the client can send let (mut client, mut server) = connect_with_connection_parameters(
ConnectionParameters::default().max_data(MIN_INITIAL_PACKET_SIZE.try_into().unwrap()),
);
let request_stream_id = make_request_and_exchange_pkts(&mut client, &mut server, false); let data_writable = |e| matches!(e, Http3ClientEvent::DataWritable { .. });
assert!(client.events().any(data_writable)); // Send a lot of data to reach the flow control limit
client.send_data(request_stream_id, &[0; 2000]).unwrap();
// now queue a priority_update packet for that stream
assert!(client
.priority_update(request_stream_id, Priority::new(6, false))
.unwrap());
let md_before = server.conn.stats().frame_tx.max_data;
// sending the http request and most most of the request data let out = client.process_output(now()); let out = server.conn.process(out.dgram(), now());
// the server responses with an ack, but the max_data didn't change
assert_eq!(md_before, server.conn.stats().frame_tx.max_data);
let out = client.process(out.dgram(), now()); let out = server.conn.process(out.dgram(), now());
// the server increased the max_data during the second read if that isn't the case // in the future and therefore this asserts fails, the request data on stream 0 could be // read to cause a max_update frame
assert_eq!(md_before + 1, server.conn.stats().frame_tx.max_data);
// make sure that the server didn't receive a priority_update on client control stream // (stream_id 2) yet letmut buf = [0; 32];
assert_eq!(
server.conn.stream_recv(StreamId::new(2), &mut buf),
Ok((0, false))
);
// the client now sends the priority update let out = client.process(out.dgram(), now());
server.conn.process_input(out.dgram().unwrap(), now());
// check that the priority_update arrived at the client control stream let num_read = server.conn.stream_recv(StreamId::new(2), &mut buf).unwrap();
assert_eq!(b"\x80\x0f\x07\x00\x04\x00\x75\x3d\x36", &buf[0..num_read.0]);
}
// Send the encoder instructions, but delay them so that the stream is blocked on decoding // headers. let encoder_inst_pkt = server.conn.process_output(now());
// Stream has been reset because of the 101 response. let e = client.events().next().unwrap();
assert_eq!(
e,
Http3ClientEvent::Reset {
stream_id: request_stream_id,
error: Error::InvalidHeader.code(),
local: true,
}
);
}
}
Messung V0.5 in Prozent
¤ Diese beiden folgenden Angebotsgruppen bietet das Unternehmen0.212Angebot
(Wie Sie bei der Firma Beratungs- und Dienstleistungen beauftragen können 2026-06-20)
¤
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.