/// States: /// - `Initializing`: this is the state during the QUIC handshake, /// - `ZeroRtt`: 0-RTT has been enabled and is active /// - Connected /// - GoingAway(StreamId): The connection has received a `GOAWAY` frame /// - Closing(CloseReason): The connection is closed. The closing has been initiated by this end of /// the connection, e.g., the `CONNECTION_CLOSE` frame has been sent. In this state, the /// connection waits a certain amount of time to retransmit the `CONNECTION_CLOSE` frame if /// needed. /// - Closed(CloseReason): This is the final close state: closing has been initialized by the peer /// and an ack for the `CONNECTION_CLOSE` frame has been sent or the closing has been initiated by /// this end of the connection and the ack for the `CONNECTION_CLOSE` has been received or the /// waiting time has passed. #[derive(Debug, PartialEq, PartialOrd, Ord, Eq, Clone)] pubenum Http3State {
Initializing,
ZeroRtt,
Connected,
GoingAway(StreamId),
Closing(CloseReason),
Closed(CloseReason),
}
/// This function is called when a not default feature needs to be negotiated. This is currently /// only used for the `WebTransport` feature. The negotiation is done via the `SETTINGS` frame /// and when the peer's `SETTINGS` frame has been received the listener will be called. pubfn set_features_listener(&mutself, feature_listener: Http3ClientEvents) { self.webtransport.set_listener(feature_listener);
}
/// This function creates and initializes, i.e. send stream type, the control and qpack /// streams. fn initialize_http3_connection(&mutself, conn: &mut Connection) -> Res<()> {
qdebug!([self], "Initialize the http3 connection."); self.control_stream_local.create(conn)?;
/// Inform an [`Http3Connection`] that a stream has data to send and that /// [`SendStream::send`] should be called for the stream. pubfn stream_has_pending_data(&mutself, stream_id: StreamId) { self.streams_with_pending_data.insert(stream_id);
}
/// Return true if there is a stream that needs to send data. pubfn has_data_to_send(&self) -> bool {
!self.streams_with_pending_data.is_empty()
}
/// This function calls the `send` function for all streams that have data to send. If a stream /// has data to send it will be added to the `streams_with_pending_data` list. /// /// Control and QPACK streams are handled differently and are never added to the list. fn send_non_control_streams(&mutself, conn: &mut Connection) -> Res<()> { let to_send = mem::take(&mutself.streams_with_pending_data); for stream_id in to_send { let done = iflet Some(s) = &mutself.send_streams.get_mut(&stream_id) {
s.send(conn)?; if s.has_data_to_send() { self.streams_with_pending_data.insert(stream_id);
}
s.done()
} else { false
}; if done { self.remove_send_stream(stream_id, conn);
}
}
Ok(())
}
/// Call `send` for all streams that need to send data. See explanation for the main structure /// for more details. pubfn process_sending(&mutself, conn: &mut Connection) -> Res<()> { // check if control stream has data to send. self.control_stream_local
.send(conn, &mutself.recv_streams)?;
/// We have a resumption token which remembers previous settings. Update the setting. pubfn set_0rtt_settings(&mutself, conn: &mut Connection, settings: HSettings) -> Res<()> { self.initialize_http3_connection(conn)?; self.set_qpack_settings(&settings)?; self.settings_state = Http3RemoteSettingsState::ZeroRtt(settings); self.state = Http3State::ZeroRtt;
Ok(())
}
/// Returns the settings for a connection. This is used for creating a resumption token. pubfn get_settings(&self) -> Option<HSettings> { iflet Http3RemoteSettingsState::Received(settings) = &self.settings_state {
Some(settings.clone())
} else {
None
}
}
/// This is called when a `ConnectionEvent::NewStream` event is received. This register the /// stream with a `NewStreamHeadReader` handler. pubfn add_new_stream(&mutself, stream_id: StreamId) {
qtrace!([self], "A new stream: {}.", stream_id); self.recv_streams.insert(
stream_id, Box::new(NewStreamHeadReader::new(stream_id, self.role)),
);
}
/// The function calls `receive` for a stream. It also deals with the outcome of a read by /// calling `handle_stream_manipulation_output`. fn stream_receive(&mutself, conn: &mut Connection, stream_id: StreamId) -> Res<ReceiveOutput> {
qtrace!([self], "Readable stream {}.", stream_id);
iflet Some(recv_stream) = self.recv_streams.get_mut(&stream_id) { let res = recv_stream.receive(conn); returnself
.handle_stream_manipulation_output(res, stream_id, conn)
.map(|(output, _)| output);
}
Ok(ReceiveOutput::NoOutput)
}
fn handle_unblocked_streams(
&mutself,
unblocked_streams: Vec<StreamId>,
conn: &mut Connection,
) -> Res<()> { for stream_id in unblocked_streams {
qdebug!([self], "Stream {} is unblocked", stream_id); iflet Some(r) = self.recv_streams.get_mut(&stream_id) { let res = r
.http_stream()
.ok_or(Error::HttpInternal(10))?
.header_unblocked(conn); let res = self.handle_stream_manipulation_output(res, stream_id, conn)?;
debug_assert!(matches!(res, (ReceiveOutput::NoOutput, _)));
}
}
Ok(())
}
/// This function handles reading from all streams, i.e. control, qpack, request/response /// stream and unidi stream that are still do not have a type. /// The function cannot handle: /// 1) a `Push(_)`, `Htttp` or `WebTransportStream(_)` stream /// 2) frames `MaxPushId`, `PriorityUpdateRequest`, `PriorityUpdateRequestPush` or `Goaway` must /// be handled by `Http3Client`/`Server`. /// /// The function returns `ReceiveOutput`. pubfn handle_stream_readable(
&mutself,
conn: &mut Connection,
stream_id: StreamId,
) -> Res<ReceiveOutput> { letmut output = self.stream_receive(conn, stream_id)?;
#[allow(clippy::match_same_arms)] // clippy is being stupid here match output {
ReceiveOutput::UnblockedStreams(unblocked_streams) => { self.handle_unblocked_streams(unblocked_streams, conn)?;
Ok(ReceiveOutput::NoOutput)
}
ReceiveOutput::ControlFrames(control_frames) => { letmut rest = Vec::new(); for cf in control_frames { iflet Some(not_handled) = self.handle_control_frame(cf)? {
rest.push(not_handled);
}
}
Ok(ReceiveOutput::ControlFrames(rest))
}
ReceiveOutput::NewStream(
NewStreamType::Push(_)
| NewStreamType::Http(_)
| NewStreamType::WebTransportStream(_),
) => Ok(output),
ReceiveOutput::NewStream(_) => {
unreachable!("NewStream should have been handled already")
}
ReceiveOutput::NoOutput => Ok(output),
}
}
/// This is called when a RESET frame has been received. pubfn handle_stream_reset(
&mutself,
stream_id: StreamId,
app_error: AppError,
conn: &mut Connection,
) -> Res<()> {
qinfo!(
[self], "Handle a stream reset stream_id={} app_err={}",
stream_id,
app_error
);
/// If the new stream is a control or QPACK stream, this function creates a proper handler /// and perform a read. /// if the new stream is a `Push(_)`, `Http` or `WebTransportStream(_)` stream, the function /// returns `ReceiveOutput::NewStream(_)` and the caller will handle it. /// If the stream is of a unknown type the stream will be closed. fn handle_new_stream(
&mutself,
conn: &mut Connection,
stream_type: NewStreamType,
stream_id: StreamId,
) -> Res<ReceiveOutput> { match stream_type {
NewStreamType::Control => { self.check_stream_exists(Http3StreamType::Control)?; self.recv_streams
.insert(stream_id, Box::new(ControlStreamRemote::new(stream_id)));
}
NewStreamType::Push(push_id) => {
qinfo!(
[self], "A new push stream {} push_id:{}.",
stream_id,
push_id
);
}
NewStreamType::Decoder => {
qdebug!([self], "A new remote qpack encoder stream {}", stream_id); self.check_stream_exists(Http3StreamType::Decoder)?; self.recv_streams.insert(
stream_id, Box::new(DecoderRecvStream::new(
stream_id,
Rc::clone(&self.qpack_decoder),
)),
);
}
NewStreamType::Encoder => {
qdebug!([self], "A new remote qpack decoder stream {}", stream_id); self.check_stream_exists(Http3StreamType::Encoder)?; self.recv_streams.insert(
stream_id, Box::new(EncoderRecvStream::new(
stream_id,
Rc::clone(&self.qpack_encoder),
)),
);
}
NewStreamType::Http(_) => {
qinfo!([self], "A new http stream {}.", stream_id);
}
NewStreamType::WebTransportStream(session_id) => { let session_exists = self
.send_streams
.get(&StreamId::from(session_id))
.is_some_and(|s| s.stream_type() == Http3StreamType::ExtendedConnect); if !session_exists {
conn.stream_stop_sending(stream_id, Error::HttpStreamCreation.code())?; return Ok(ReceiveOutput::NoOutput);
} // set incoming WebTransport streams to be fair (share bandwidth)
conn.stream_fairness(stream_id, true).ok();
qinfo!(
[self], "A new WebTransport stream {} for session {}.",
stream_id,
session_id
);
}
NewStreamType::Unknown => {
conn.stream_stop_sending(stream_id, Error::HttpStreamCreation.code())?;
}
};
/// This is called when an application closes the connection. pubfn close(&mutself, error: AppError) {
qdebug!([self], "Close connection error {:?}.", error); self.state = Http3State::Closing(CloseReason::Application(error)); if (!self.send_streams.is_empty() || !self.recv_streams.is_empty()) && (error == 0) {
qwarn!("close(0) called when streams still active");
} self.send_streams.clear(); self.recv_streams.clear();
}
/// This function will not handle the output of the function completely, but only /// handle the indication that a stream is closed. There are 2 cases: /// - an error occurred or /// - the stream is done, i.e. the second value in `output` tuple is true if the stream is done /// and can be removed from the `recv_streams` /// /// How it is handling `output`: /// - if the stream is done, it removes the stream from `recv_streams` /// - if the stream is not done and there is no error, return `output` and the caller will /// handle it. /// - in case of an error: /// - if it is only a stream error and the stream is not critical, send `STOP_SENDING` frame, /// remove the stream from `recv_streams` and inform the listener that the stream has been /// reset. /// - otherwise this is a connection error. In this case, propagate the error to the caller /// that will handle it properly. fn handle_stream_manipulation_output<U>(
&mutself,
output: Res<(U, bool)>,
stream_id: StreamId,
conn: &mut Connection,
) -> Res<(U, bool)> where
U: Default,
{ match &output {
Ok((_, true)) => { self.remove_recv_stream(stream_id, conn);
}
Ok((_, false)) => {}
Err(e) => { if e.stream_reset_error() && !self.recv_stream_is_critical(stream_id) {
mem::drop(conn.stream_stop_sending(stream_id, e.code())); self.close_recv(stream_id, CloseType::LocalError(e.code()), conn)?; return Ok((U::default(), false));
}
}
}
output
}
// Call immediately send so that at least headers get sent. This will make Firefox faster, // since it can send request body immediately in most cases and does not need to do // a complete process loop. self.send_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?
.send(conn)?;
Ok(())
}
/// Stream 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 happens while reading a stream, /// e.g. early close, protocol error, etc. pubfn read_data(
&mutself,
conn: &mut Connection,
stream_id: StreamId,
buf: &mut [u8],
) -> Res<(usize, bool)> {
qdebug!([self], "read_data from stream {}.", stream_id); let res = self
.recv_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?
.read_data(conn, buf); self.handle_stream_manipulation_output(res, stream_id, conn)
}
/// This is called when an application resets a stream. /// The application reset will close both sides. pubfn stream_reset_send(
&mutself,
conn: &mut Connection,
stream_id: StreamId,
error: AppError,
) -> Res<()> {
qinfo!(
[self], "Reset sending side of stream {} error={}.",
stream_id,
error
);
// Stream may be already be closed and we may get an error here, but we do not care.
conn.stream_stop_sending(stream_id, error)?;
Ok(())
}
/// Set the stream `SendOrder`. /// /// # Errors /// /// Returns `InvalidStreamId` if the stream id doesn't exist pubfn stream_set_sendorder(
conn: &mut Connection,
stream_id: StreamId,
sendorder: Option<SendOrder>,
) -> Res<()> {
conn.stream_sendorder(stream_id, sendorder)
.map_err(|_| Error::InvalidStreamId)
}
/// Set the stream Fairness. Fair streams will share bandwidth with other /// streams of the same sendOrder group (or the unordered group). Unfair streams /// will give bandwidth preferentially to the lowest streamId with data to send. /// /// # Errors /// /// Returns `InvalidStreamId` if the stream id doesn't exist pubfn stream_set_fairness(
conn: &mut Connection,
stream_id: StreamId,
fairness: bool,
) -> Res<()> {
conn.stream_fairness(stream_id, fairness)
.map_err(|_| Error::InvalidStreamId)
}
pubfn cancel_fetch(
&mutself,
stream_id: StreamId,
error: AppError,
conn: &mut Connection,
) -> Res<()> {
qinfo!([self], "cancel_fetch {} error={}.", stream_id, error); let send_stream = self.send_streams.get(&stream_id); let recv_stream = self.recv_streams.get(&stream_id); match (send_stream, recv_stream) {
(None, None) => return Err(Error::InvalidStreamId),
(Some(s), None) => { if !matches!(
s.stream_type(),
Http3StreamType::Http | Http3StreamType::ExtendedConnect
) { return Err(Error::InvalidStreamId);
} // Stream may be already be closed and we may get an error here, but we do not care.
mem::drop(self.stream_reset_send(conn, stream_id, error));
}
(None, Some(s)) => { if !matches!(
s.stream_type(),
Http3StreamType::Http
| Http3StreamType::Push
| Http3StreamType::ExtendedConnect
) { return Err(Error::InvalidStreamId);
}
// Stream may be already be closed and we may get an error here, but we do not care.
mem::drop(self.stream_stop_sending(conn, stream_id, error));
}
(Some(s), Some(r)) => {
debug_assert_eq!(s.stream_type(), r.stream_type()); if !matches!(
s.stream_type(),
Http3StreamType::Http | Http3StreamType::ExtendedConnect
) { return Err(Error::InvalidStreamId);
} // Stream may be already be closed and we may get an error here, but we do not care.
mem::drop(self.stream_reset_send(conn, stream_id, error)); // Stream may be already be closed and we may get an error here, but we do not care.
mem::drop(self.stream_stop_sending(conn, stream_id, error));
}
}
Ok(())
}
/// This is called when an application wants to close the sending side of a stream. pubfn stream_close_send(&mutself, conn: &mut Connection, stream_id: StreamId) -> Res<()> {
qdebug!([self], "Close the sending side for stream {}.", stream_id);
debug_assert!(self.state.active()); let send_stream = self
.send_streams
.get_mut(&stream_id)
.ok_or(Error::InvalidStreamId)?; // The following function may return InvalidStreamId from the transport layer if the stream // has been closed already. It is ok to ignore it here.
mem::drop(send_stream.close(conn)); if send_stream.done() { self.remove_send_stream(stream_id, conn);
} elseif send_stream.has_data_to_send() { self.streams_with_pending_data.insert(stream_id);
}
Ok(())
}
let wt = self
.recv_streams
.get(&session_id)
.ok_or(Error::InvalidStreamId)?
.webtransport()
.ok_or(Error::InvalidStreamId)?; if !wt.borrow().is_active() { return Err(Error::InvalidStreamId);
}
let stream_id = conn
.stream_create(stream_type)
.map_err(|e| Error::map_stream_create_errors(&e))?; // Set outgoing WebTransport streams to be fair (share bandwidth) // This really can't fail, panics if it does
conn.stream_fairness(stream_id, true).unwrap();
/// If the control stream has received frames `MaxPushId`, `Goaway`, `PriorityUpdateRequest` or /// `PriorityUpdateRequestPush` which handling is specific to the client and server, we must /// give them to the specific client/server handler. fn handle_control_frame(&mutself, f: HFrame) -> Res<Option<HFrame>> {
qdebug!([self], "Handle a control frame {:?}", f); if !matches!(f, HFrame::Settings { .. })
&& !matches!( self.settings_state,
Http3RemoteSettingsState::Received { .. }
)
{ return Err(Error::HttpMissingSettings);
} match f {
HFrame::Settings { settings } => { self.handle_settings(settings)?;
Ok(None)
}
HFrame::Goaway { .. }
| HFrame::MaxPushId { .. }
| HFrame::CancelPush { .. }
| HFrame::PriorityUpdateRequest { .. }
| HFrame::PriorityUpdatePush { .. } => Ok(Some(f)),
_ => Err(Error::HttpFrameUnexpected),
}
}
fn handle_settings(&mutself, new_settings: HSettings) -> Res<()> {
qdebug!([self], "Handle SETTINGS frame."); match &self.settings_state {
Http3RemoteSettingsState::NotReceived => { self.set_qpack_settings(&new_settings)?; self.webtransport.handle_settings(&new_settings); self.settings_state = Http3RemoteSettingsState::Received(new_settings);
Ok(())
}
Http3RemoteSettingsState::ZeroRtt(settings) => { self.webtransport.handle_settings(&new_settings); letmut qpack_changed = false; for st in &[
HSettingType::MaxHeaderListSize,
HSettingType::MaxTableCapacity,
HSettingType::BlockedStreams,
] { let zero_rtt_value = settings.get(*st); let new_value = new_settings.get(*st); if zero_rtt_value == new_value { continue;
} if zero_rtt_value > new_value {
qerror!(
[self], "The new({}) and the old value({}) of setting {:?} do not match",
new_value,
zero_rtt_value,
st
); return Err(Error::HttpSettings);
}
match st {
HSettingType::MaxTableCapacity => { if zero_rtt_value != 0 { return Err(Error::QpackError(neqo_qpack::Error::DecoderStream));
}
qpack_changed = true;
}
HSettingType::BlockedStreams => qpack_changed = true,
HSettingType::MaxHeaderListSize
| HSettingType::EnableWebTransport
| HSettingType::EnableH3Datagram => (),
}
} if qpack_changed {
qdebug!([self], "Settings after zero rtt differ."); self.set_qpack_settings(&(new_settings))?;
} self.settings_state = Http3RemoteSettingsState::Received(new_settings);
Ok(())
}
Http3RemoteSettingsState::Received { .. } => Err(Error::HttpFrameUnexpected),
}
}
/// Return the current state on `Http3Connection`. pubfn state(&self) -> Http3State { self.state.clone()
}
/// Adds a new send and receive stream. pubfn add_streams(
&mutself,
stream_id: StreamId,
send_stream: Box<dyn SendStream>,
recv_stream: Box<dyn RecvStream>,
) { if send_stream.has_data_to_send() { self.streams_with_pending_data.insert(stream_id);
} self.send_streams.insert(stream_id, send_stream); self.recv_streams.insert(stream_id, recv_stream);
}
/// Add a new recv stream. This is used for push streams. pubfn add_recv_stream(&mutself, stream_id: StreamId, recv_stream: Box<dyn RecvStream>) { self.recv_streams.insert(stream_id, recv_stream);
}
for id in recv {
qtrace!("Remove the extended connect sub receiver stream {}", id); // Use CloseType::ResetRemote so that an event will be sent. CloseType::LocalError would // have the same effect. iflet Some(mut s) = self.recv_streams.remove(&id) {
mem::drop(s.reset(CloseType::ResetRemote(Error::HttpRequestCancelled.code())));
}
mem::drop(conn.stream_stop_sending(id, Error::HttpRequestCancelled.code()));
} for id in send {
qtrace!("Remove the extended connect sub send stream {}", id); iflet Some(mut s) = self.send_streams.remove(&id) {
s.handle_stop_sending(CloseType::ResetRemote(Error::HttpRequestCancelled.code()));
}
mem::drop(conn.stream_reset_send(id, Error::HttpRequestCancelled.code()));
}
}
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.