h3i/actions/
h3.rs

1// Copyright (C) 2024, Cloudflare, Inc.
2// All rights reserved.
3//
4// Redistribution and use in source and binary forms, with or without
5// modification, are permitted provided that the following conditions are
6// met:
7//
8//     * Redistributions of source code must retain the above copyright notice,
9//       this list of conditions and the following disclaimer.
10//
11//     * Redistributions in binary form must reproduce the above copyright
12//       notice, this list of conditions and the following disclaimer in the
13//       documentation and/or other materials provided with the distribution.
14//
15// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
16// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
17// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
19// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
22// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
23// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
24// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
25// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26
27//! Actions specific to HTTP/3 and QUIC
28//!
29//! Actions are small operations such as sending HTTP/3 frames or managing QUIC
30//! streams. Each independent use case for h3i requires its own collection of
31//! Actions, that h3i iterates over in sequence and executes.
32
33use std::collections::HashMap;
34use std::time::Duration;
35
36use quiche;
37use quiche::h3::frame::Frame;
38use quiche::h3::Header;
39use quiche::ConnectionError;
40use serde::Deserialize;
41use serde::Serialize;
42use serde_with::serde_as;
43
44use crate::encode_header_block;
45use crate::encode_header_block_literal;
46
47/// An action which the HTTP/3 client should take.
48///
49/// The client iterates over a vector of said actions, executing each one
50/// sequentially. Note that packets will be flushed when said iteration has
51/// completed, regardless of if an [`Action::FlushPackets`] was the terminal
52/// action.
53#[derive(Clone, Debug, PartialEq, Eq)]
54pub enum Action {
55    /// Send a [quiche::h3::frame::Frame] over a stream.
56    SendFrame {
57        stream_id: u64,
58        fin_stream: bool,
59        frame: Frame,
60    },
61
62    /// Send a HEADERS frame over a stream.
63    SendHeadersFrame {
64        stream_id: u64,
65        fin_stream: bool,
66        literal_headers: bool,
67        headers: Vec<Header>,
68        frame: Frame,
69    },
70
71    /// Send arbitrary bytes over a stream.
72    StreamBytes {
73        stream_id: u64,
74        fin_stream: bool,
75        bytes: Vec<u8>,
76    },
77
78    /// Open a new unidirectional stream.
79    OpenUniStream {
80        stream_id: u64,
81        fin_stream: bool,
82        stream_type: u64,
83    },
84
85    /// Send a RESET_STREAM frame with the given error code.
86    ResetStream {
87        stream_id: u64,
88        error_code: u64,
89    },
90
91    /// Send a STOP_SENDING frame with the given error code.
92    StopSending {
93        stream_id: u64,
94        error_code: u64,
95    },
96
97    /// Send a CONNECTION_CLOSE frame with the given [`ConnectionError`].
98    ConnectionClose {
99        error: ConnectionError,
100    },
101
102    FlushPackets,
103
104    /// Wait for an event. See [WaitType] for the events.
105    Wait {
106        wait_type: WaitType,
107    },
108}
109
110/// Configure the wait behavior for a connection.
111#[serde_as]
112#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
113pub enum WaitType {
114    /// Wait for a time before firing the next action
115    #[serde(rename = "duration")]
116    WaitDuration(
117        #[serde_as(as = "serde_with::DurationMilliSecondsWithFrac<f64>")]
118        Duration,
119    ),
120    /// Wait for some form of a response before firing the next action. This can
121    /// be superseded in several cases:
122    /// 1. The peer resets the specified stream.
123    /// 2. The peer sends a `fin` over the specified stream
124    StreamEvent(StreamEvent),
125}
126
127impl From<WaitType> for Action {
128    fn from(value: WaitType) -> Self {
129        Self::Wait { wait_type: value }
130    }
131}
132
133/// A response event, received over a stream, which will terminate the wait
134/// period.
135///
136/// See [StreamEventType] for the types of events.
137#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
138#[serde(rename = "snake_case")]
139pub struct StreamEvent {
140    pub stream_id: u64,
141    #[serde(rename = "type")]
142    pub event_type: StreamEventType,
143}
144
145/// Response that can terminate a wait period.
146#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
147#[serde(rename_all = "lowercase")]
148pub enum StreamEventType {
149    /// A HEADERS frame was received.
150    Headers,
151    /// A DATA frame was received.
152    Data,
153    /// The stream was somehow finished, either by a RESET_STREAM frame or via
154    /// the `fin` bit being set.
155    Finished,
156}
157
158#[derive(Debug, Default)]
159pub(crate) struct WaitingFor(HashMap<u64, Vec<StreamEvent>>);
160
161impl WaitingFor {
162    pub(crate) fn is_empty(&self) -> bool {
163        self.0.values().all(|v| v.is_empty())
164    }
165
166    pub(crate) fn add_wait(&mut self, stream_event: &StreamEvent) {
167        self.0
168            .entry(stream_event.stream_id)
169            .or_default()
170            .push(*stream_event);
171    }
172
173    pub(crate) fn remove_wait(&mut self, stream_event: StreamEvent) {
174        if let Some(waits) = self.0.get_mut(&stream_event.stream_id) {
175            let old_len = waits.len();
176            waits.retain(|wait| wait != &stream_event);
177            let new_len = waits.len();
178
179            if old_len != new_len {
180                log::info!("No longer waiting for {:?}", stream_event);
181            }
182        }
183    }
184
185    pub(crate) fn clear_waits_on_stream(&mut self, stream_id: u64) {
186        if let Some(waits) = self.0.get_mut(&stream_id) {
187            if !waits.is_empty() {
188                log::info!("Clearing all waits for stream {}", stream_id);
189                waits.clear();
190            }
191        }
192    }
193}
194
195/// Convenience to convert between header-related data and a
196/// [Action::SendHeadersFrame].
197pub fn send_headers_frame(
198    stream_id: u64, fin_stream: bool, headers: Vec<Header>,
199) -> Action {
200    let header_block = encode_header_block(&headers).unwrap();
201
202    Action::SendHeadersFrame {
203        stream_id,
204        fin_stream,
205        headers,
206        literal_headers: false,
207        frame: Frame::Headers { header_block },
208    }
209}
210
211/// Convenience to convert between header-related data and a
212/// [Action::SendHeadersFrame]. Unlike [`send_headers_frame`],
213/// this version encodes the headers literally as they are provided,
214/// not converting the header names to lower-case.
215pub fn send_headers_frame_literal(
216    stream_id: u64, fin_stream: bool, headers: Vec<Header>,
217) -> Action {
218    let header_block = encode_header_block_literal(&headers).unwrap();
219
220    Action::SendHeadersFrame {
221        stream_id,
222        fin_stream,
223        headers,
224        literal_headers: true,
225        frame: Frame::Headers { header_block },
226    }
227}
h3i/actions/h3.rs

h3i/actions/
h3.rs
