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;
45
46/// An action which the HTTP/3 client should take.
47///
48/// The client iterates over a vector of said actions, executing each one
49/// sequentially. Note that packets will be flushed when said iteration has
50/// completed, regardless of if an [`Action::FlushPackets`] was the terminal
51/// action.
52#[derive(Clone, Debug, PartialEq, Eq)]
53pub enum Action {
54    /// Send a [quiche::h3::frame::Frame] over a stream.
55    SendFrame {
56        stream_id: u64,
57        fin_stream: bool,
58        frame: Frame,
59    },
60
61    /// Send a HEADERS frame over a stream.
62    SendHeadersFrame {
63        stream_id: u64,
64        fin_stream: bool,
65        headers: Vec<Header>,
66        frame: Frame,
67    },
68
69    /// Send arbitrary bytes over a stream.
70    StreamBytes {
71        stream_id: u64,
72        fin_stream: bool,
73        bytes: Vec<u8>,
74    },
75
76    /// Open a new unidirectional stream.
77    OpenUniStream {
78        stream_id: u64,
79        fin_stream: bool,
80        stream_type: u64,
81    },
82
83    /// Send a RESET_STREAM frame with the given error code.
84    ResetStream {
85        stream_id: u64,
86        error_code: u64,
87    },
88
89    /// Send a STOP_SENDING frame with the given error code.
90    StopSending {
91        stream_id: u64,
92        error_code: u64,
93    },
94
95    /// Send a CONNECTION_CLOSE frame with the given [`ConnectionError`].
96    ConnectionClose {
97        error: ConnectionError,
98    },
99
100    FlushPackets,
101
102    /// Wait for an event. See [WaitType] for the events.
103    Wait {
104        wait_type: WaitType,
105    },
106}
107
108/// Configure the wait behavior for a connection.
109#[serde_as]
110#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
111pub enum WaitType {
112    /// Wait for a time before firing the next action
113    #[serde(rename = "duration")]
114    WaitDuration(
115        #[serde_as(as = "serde_with::DurationMilliSecondsWithFrac<f64>")]
116        Duration,
117    ),
118    /// Wait for some form of a response before firing the next action. This can
119    /// be superseded in several cases:
120    /// 1. The peer resets the specified stream.
121    /// 2. The peer sends a `fin` over the specified stream
122    StreamEvent(StreamEvent),
123}
124
125impl From<WaitType> for Action {
126    fn from(value: WaitType) -> Self {
127        Self::Wait { wait_type: value }
128    }
129}
130
131/// A response event, received over a stream, which will terminate the wait
132/// period.
133///
134/// See [StreamEventType] for the types of events.
135#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
136#[serde(rename = "snake_case")]
137pub struct StreamEvent {
138    pub stream_id: u64,
139    #[serde(rename = "type")]
140    pub event_type: StreamEventType,
141}
142
143/// Response that can terminate a wait period.
144#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
145#[serde(rename_all = "lowercase")]
146pub enum StreamEventType {
147    /// A HEADERS frame was received.
148    Headers,
149    /// A DATA frame was received.
150    Data,
151    /// The stream was somehow finished, either by a RESET_STREAM frame or via
152    /// the `fin` bit being set.
153    Finished,
154}
155
156#[derive(Debug, Default)]
157pub(crate) struct WaitingFor(HashMap<u64, Vec<StreamEvent>>);
158
159impl WaitingFor {
160    pub(crate) fn is_empty(&self) -> bool {
161        self.0.values().all(|v| v.is_empty())
162    }
163
164    pub(crate) fn add_wait(&mut self, stream_event: &StreamEvent) {
165        self.0
166            .entry(stream_event.stream_id)
167            .or_default()
168            .push(*stream_event);
169    }
170
171    pub(crate) fn remove_wait(&mut self, stream_event: StreamEvent) {
172        if let Some(waits) = self.0.get_mut(&stream_event.stream_id) {
173            let old_len = waits.len();
174            waits.retain(|wait| wait != &stream_event);
175            let new_len = waits.len();
176
177            if old_len != new_len {
178                log::info!("No longer waiting for {:?}", stream_event);
179            }
180        }
181    }
182
183    pub(crate) fn clear_waits_on_stream(&mut self, stream_id: u64) {
184        if let Some(waits) = self.0.get_mut(&stream_id) {
185            if !waits.is_empty() {
186                log::info!("Clearing all waits for stream {}", stream_id);
187                waits.clear();
188            }
189        }
190    }
191}
192
193/// Convenience to convert between header-related data and a
194/// [Action::SendHeadersFrame].
195pub fn send_headers_frame(
196    stream_id: u64, fin_stream: bool, headers: Vec<Header>,
197) -> Action {
198    let header_block = encode_header_block(&headers).unwrap();
199
200    Action::SendHeadersFrame {
201        stream_id,
202        fin_stream,
203        headers,
204        frame: Frame::Headers { header_block },
205    }
206}
h3i/actions/h3.rs

h3i/actions/
h3.rs
