1use super::header::{
2 get_header, is_json_media_type, matches_media_type, normalize_header_name, normalize_headers,
3 parse_content_length, strict_content_length_values, validate_header_name,
4 validate_header_value,
5};
6use crate::{BootError, Result, SseEvent, SseStream};
7use futures_core::Stream;
8use serde::de::DeserializeOwned;
9use serde::Serialize;
10use std::collections::BTreeMap;
11use std::fmt;
12use std::sync::{Arc, Mutex};
13
14#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct BootResponse {
17 pub status: u16,
18 pub headers: BTreeMap<String, String>,
19 pub appended_headers: Vec<(String, String)>,
20 pub body: Vec<u8>,
21 stream: Option<SharedSseStream>,
22}
23
24#[derive(Clone)]
25struct SharedSseStream {
26 inner: Arc<Mutex<Option<SseStream>>>,
27}
28
29impl SharedSseStream {
30 fn new(stream: SseStream) -> Self {
31 Self {
32 inner: Arc::new(Mutex::new(Some(stream))),
33 }
34 }
35
36 fn take(&self) -> Option<SseStream> {
37 self.inner.lock().ok()?.take()
38 }
39}
40
41impl fmt::Debug for SharedSseStream {
42 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43 f.debug_struct("SharedSseStream").finish_non_exhaustive()
44 }
45}
46
47impl PartialEq for SharedSseStream {
48 fn eq(&self, other: &Self) -> bool {
49 Arc::ptr_eq(&self.inner, &other.inner)
50 }
51}
52
53impl Eq for SharedSseStream {}
54
55impl Default for BootResponse {
56 fn default() -> Self {
57 Self::new(200, Vec::<u8>::new())
58 }
59}
60
61impl BootResponse {
62 pub fn new(status: u16, body: impl Into<Vec<u8>>) -> Self {
63 Self {
64 status,
65 headers: BTreeMap::new(),
66 appended_headers: Vec::new(),
67 body: body.into(),
68 stream: None,
69 }
70 }
71
72 pub fn status(&self) -> u16 {
73 self.status
74 }
75
76 pub fn body(&self) -> &[u8] {
77 &self.body
78 }
79
80 pub fn into_body(self) -> Vec<u8> {
81 self.body
82 }
83
84 pub fn empty(status: u16) -> Self {
85 Self::new(status, Vec::<u8>::new())
86 }
87
88 pub fn no_content() -> Self {
89 Self::empty(204)
90 }
91
92 pub fn redirect(location: impl Into<String>) -> Self {
93 Self::redirect_with_status(302, location)
94 }
95
96 pub fn see_other(location: impl Into<String>) -> Self {
97 Self::redirect_with_status(303, location)
98 }
99
100 pub fn temporary_redirect(location: impl Into<String>) -> Self {
101 Self::redirect_with_status(307, location)
102 }
103
104 pub fn permanent_redirect(location: impl Into<String>) -> Self {
105 Self::redirect_with_status(308, location)
106 }
107
108 pub fn redirect_with_status(status: u16, location: impl Into<String>) -> Self {
109 Self::empty(status).with_location(location)
110 }
111
112 pub fn text(body: impl Into<String>) -> Self {
113 Self::text_with_status(200, body)
114 }
115
116 pub fn text_with_status(status: u16, body: impl Into<String>) -> Self {
117 Self::new(status, body.into()).with_header("content-type", "text/plain; charset=utf-8")
118 }
119
120 pub fn json<T>(body: &T) -> Result<Self>
121 where
122 T: Serialize,
123 {
124 Self::json_with_status(200, body)
125 }
126
127 pub fn json_with_status<T>(status: u16, body: &T) -> Result<Self>
128 where
129 T: Serialize,
130 {
131 let body = serde_json::to_vec(body).map_err(|err| BootError::Internal(err.to_string()))?;
132 Ok(Self::new(status, body).with_header("content-type", "application/json"))
133 }
134
135 pub fn sse<S>(stream: S) -> Self
136 where
137 S: Stream<Item = Result<SseEvent>> + Send + 'static,
138 {
139 Self::empty(200)
140 .with_header("content-type", "text/event-stream; charset=utf-8")
141 .with_header("cache-control", "no-cache")
142 .with_header("connection", "keep-alive")
143 .with_sse_stream(stream)
144 }
145
146 pub fn from_error(error: &BootError) -> Self {
147 Self::text_with_status(error.http_status_code(), error.http_response_message())
148 }
149
150 pub fn body_text(&self) -> Result<String> {
151 if self.is_streaming() {
152 return Err(BootError::Internal(
153 "streaming response body cannot be read as text".to_string(),
154 ));
155 }
156 String::from_utf8(self.body.clone()).map_err(|err| BootError::Internal(err.to_string()))
157 }
158
159 pub fn body_json<T>(&self) -> Result<T>
160 where
161 T: DeserializeOwned,
162 {
163 if self.is_streaming() {
164 return Err(BootError::Internal(
165 "streaming response body cannot be read as JSON".to_string(),
166 ));
167 }
168 serde_json::from_slice(&self.body).map_err(|err| BootError::Internal(err.to_string()))
169 }
170
171 pub fn with_status(mut self, status: u16) -> Self {
172 self.status = status;
173 self
174 }
175
176 pub fn with_content_type(self, content_type: impl Into<String>) -> Self {
177 self.with_header("content-type", content_type)
178 }
179
180 pub fn with_content_length(self, content_length: u64) -> Self {
181 self.with_header("content-length", content_length.to_string())
182 }
183
184 pub fn with_location(self, location: impl Into<String>) -> Self {
185 self.with_header("location", location)
186 }
187
188 pub fn with_www_authenticate(self, challenge: impl Into<String>) -> Self {
189 self.with_header("www-authenticate", challenge)
190 }
191
192 pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
193 self.headers
194 .insert(normalize_header_name(name), value.into());
195 self
196 }
197
198 pub fn with_headers(mut self, headers: BTreeMap<String, String>) -> Self {
199 self.headers = normalize_headers(headers);
200 self
201 }
202
203 pub fn append_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
204 self.appended_headers
205 .push((normalize_header_name(name), value.into()));
206 self
207 }
208
209 pub fn append_www_authenticate(self, challenge: impl Into<String>) -> Self {
210 self.append_header("www-authenticate", challenge)
211 }
212
213 pub fn is_streaming(&self) -> bool {
214 self.stream.is_some()
215 }
216
217 pub fn is_event_stream(&self) -> bool {
218 self.is_content_type("text/event-stream")
219 }
220
221 pub fn into_sse_stream(self) -> Option<SseStream> {
222 self.stream.and_then(|stream| stream.take())
223 }
224
225 pub fn header_entries(&self) -> impl Iterator<Item = (&str, &str)> {
226 self.headers
227 .iter()
228 .map(|(name, value)| (name.as_str(), value.as_str()))
229 .chain(
230 self.appended_headers
231 .iter()
232 .map(|(name, value)| (name.as_str(), value.as_str())),
233 )
234 }
235
236 pub fn validate_headers(&self) -> Result<()> {
237 for (name, value) in self.header_entries() {
238 validate_response_header(name, value)?;
239 }
240
241 Ok(())
242 }
243
244 pub fn header(&self, name: &str) -> Option<&str> {
245 get_header(&self.headers, name)
246 }
247
248 pub fn header_values(&self, name: &str) -> Vec<&str> {
249 let mut values = self.header(name).into_iter().collect::<Vec<_>>();
250 values.extend(
251 self.appended_headers
252 .iter()
253 .filter(|(key, _)| key.eq_ignore_ascii_case(name))
254 .map(|(_, value)| value.as_str()),
255 );
256 values
257 }
258
259 pub fn content_type(&self) -> Option<&str> {
260 self.header_values("content-type").into_iter().next()
261 }
262
263 pub fn location(&self) -> Option<&str> {
264 self.header_values("location").into_iter().next()
265 }
266
267 pub fn www_authenticate(&self) -> Option<&str> {
268 self.header_values("www-authenticate").into_iter().next()
269 }
270
271 pub fn www_authenticate_values(&self) -> Vec<&str> {
272 self.header_values("www-authenticate")
273 }
274
275 pub fn content_length(&self) -> Result<Option<u64>> {
276 let Some(content_length) = self.header_values("content-length").into_iter().next() else {
277 return Ok(None);
278 };
279
280 parse_content_length(content_length)
281 .map(Some)
282 .ok_or_else(|| {
283 BootError::Internal(format!("invalid content-length header: {content_length}"))
284 })
285 }
286
287 pub fn strict_content_length(&self) -> Result<Option<u64>> {
288 strict_content_length_values(
289 self.header_values("content-length"),
290 |content_length| {
291 BootError::Internal(format!(
292 "invalid response content-length header: {content_length}"
293 ))
294 },
295 |expected_content_length, content_length| {
296 BootError::Internal(format!(
297 "conflicting response content-length headers: {expected_content_length} != {content_length}"
298 ))
299 },
300 )
301 }
302
303 pub fn validate_content_length(&self) -> Result<()> {
304 let Some(content_length) = self.strict_content_length()? else {
305 return Ok(());
306 };
307 if self.is_streaming() {
308 return Err(BootError::Internal(
309 "streaming responses must not include a content-length header".to_string(),
310 ));
311 }
312
313 let actual_body_length = self.body.len() as u64;
314 if actual_body_length == content_length {
315 return Ok(());
316 }
317
318 Err(BootError::Internal(format!(
319 "response content-length header does not match response body length: expected {content_length}, got {actual_body_length}"
320 )))
321 }
322
323 pub fn is_content_type(&self, media_type: &str) -> bool {
324 self.content_type()
325 .is_some_and(|content_type| matches_media_type(content_type, media_type))
326 }
327
328 pub fn is_json_content_type(&self) -> bool {
329 self.content_type().is_some_and(is_json_media_type)
330 }
331
332 pub fn has_body(&self) -> bool {
333 self.is_streaming() || !self.body.is_empty()
334 }
335
336 pub fn allows_body(&self) -> bool {
337 !(self.is_informational() || self.status == 204 || self.status == 304)
338 }
339
340 pub fn validate_body_allowed(&self) -> Result<()> {
341 if !self.has_body() || self.allows_body() {
342 return Ok(());
343 }
344
345 Err(BootError::Internal(format!(
346 "response status {} must not include a body",
347 self.status
348 )))
349 }
350
351 pub fn is_informational(&self) -> bool {
352 (100..200).contains(&self.status)
353 }
354
355 pub fn is_success(&self) -> bool {
356 (200..300).contains(&self.status)
357 }
358
359 pub fn is_redirection(&self) -> bool {
360 (300..400).contains(&self.status)
361 }
362
363 pub fn is_client_error(&self) -> bool {
364 (400..500).contains(&self.status)
365 }
366
367 pub fn is_server_error(&self) -> bool {
368 (500..600).contains(&self.status)
369 }
370
371 pub fn is_error(&self) -> bool {
372 self.is_client_error() || self.is_server_error()
373 }
374
375 pub fn is_valid_status(&self) -> bool {
376 (100..1000).contains(&self.status)
377 }
378
379 pub fn validate_status(&self) -> Result<()> {
380 if self.is_valid_status() {
381 return Ok(());
382 }
383
384 Err(BootError::Internal(format!(
385 "invalid response status {}",
386 self.status
387 )))
388 }
389
390 pub fn validate(&self) -> Result<()> {
391 self.validate_status()?;
392 self.validate_content_length()?;
393 self.validate_body_allowed()?;
394 self.validate_headers()
395 }
396
397 fn with_sse_stream<S>(mut self, stream: S) -> Self
398 where
399 S: Stream<Item = Result<SseEvent>> + Send + 'static,
400 {
401 self.stream = Some(SharedSseStream::new(Box::pin(stream)));
402 self
403 }
404}
405
406fn validate_response_header(name: &str, value: &str) -> Result<()> {
407 validate_header_name(name).map_err(|message| {
408 BootError::Internal(format!("invalid response header name {name:?}: {message}"))
409 })?;
410 validate_header_value(value).map_err(|message| {
411 BootError::Internal(format!(
412 "invalid response header value for {name:?}: {message}"
413 ))
414 })
415}