web_faith_encoding/
request.rs

1//! Applying a content coding to a request body.
2//!
3//! There is no standard signal that indicates a server accepts request body encoding ahead of
4//! sending. Therefore, doing so always requires out-of-band knowledge in some way or shape.
5//! ([RFC 9110 §15.5.16](https://www.rfc-editor.org/rfc/rfc9110#section-15.5.16) does specify that
6//! servers should answer encodings they can't decode with an `Accept-Encoding` header; that would
7//! require buffering and re-sending the request, so we don't implement it automatically.)
8
9use std::{io, pin::Pin};
10
11use async_compression::tokio::bufread::{BrotliEncoder, GzipEncoder, ZlibEncoder, ZstdEncoder};
12use bytes::Bytes;
13use futures::Stream;
14use tokio::io::AsyncReadExt;
15use tokio_util::io::{ReaderStream, StreamReader};
16
17use http::header::{CONTENT_ENCODING, CONTENT_LENGTH, HeaderMap};
18
19use crate::{Coding, ContentEncoding};
20
21/// A request body stream, as reqwest takes one.
22pub type RequestStream = Pin<Box<dyn Stream<Item = io::Result<Bytes>> + Send>>;
23
24/// Compress a request body in `coding`, and declare it in the headers.
25///
26/// Returns the bytes that go on the wire, with `Content-Encoding` naming `coding` after whatever
27/// the caller had already declared, and `Content-Length` removed: it described the body before
28/// compression. The two cannot disagree about what the body carries.
29///
30/// [`compress_buffer`] and [`ContentEncoding::layer`] are the halves, for a caller driving them
31/// separately.
32pub async fn encode(headers: &mut HeaderMap, body: &[u8], coding: Coding) -> io::Result<Vec<u8>> {
33	let compressed = compress_buffer(body, coding.clone()).await?;
34	declare(headers, coding)?;
35	Ok(compressed)
36}
37
38/// Compress a streaming request body in `coding`, and declare it in the headers.
39///
40/// As [`encode`], for a body arriving in chunks. It goes out chunked, having no length to declare.
41pub fn encode_stream<S>(
42	headers: &mut HeaderMap,
43	body: S,
44	coding: Coding,
45) -> io::Result<RequestStream>
46where
47	S: Stream<Item = io::Result<Bytes>> + Send + 'static,
48{
49	let compressed = compress_stream(body, coding.clone())?;
50	declare(headers, coding)?;
51	Ok(compressed)
52}
53
54/// Add `coding` to what the headers already declare, and drop the length it no longer describes.
55fn declare(headers: &mut HeaderMap, coding: Coding) -> io::Result<()> {
56	let layered = ContentEncoding::from(&*headers).layer(coding);
57	let value = layered
58		.to_header_value()
59		.ok_or_else(|| io::Error::other(format!("cannot declare {layered:?}")))?;
60
61	headers.insert(CONTENT_ENCODING, value);
62	headers.remove(CONTENT_LENGTH);
63	Ok(())
64}
65
66/// Compress a buffered request body, yielding the bytes that go on the wire.
67///
68/// The length of the result is the `Content-Length` the request can declare.
69// spec:ENC#what-a-compressed-request-sends
70pub async fn compress_buffer(input: &[u8], coding: Coding) -> io::Result<Vec<u8>> {
71	let mut output = Vec::new();
72	match coding {
73		Coding::Gzip => GzipEncoder::new(input).read_to_end(&mut output).await?,
74		Coding::Deflate => ZlibEncoder::new(input).read_to_end(&mut output).await?,
75		Coding::Brotli => BrotliEncoder::new(input).read_to_end(&mut output).await?,
76		Coding::Zstd => ZstdEncoder::new(input).read_to_end(&mut output).await?,
77		other => return Err(unsupported(&other)),
78	};
79	Ok(output)
80}
81
82/// The error a coding this crate cannot apply produces.
83fn unsupported(coding: &Coding) -> io::Error {
84	io::Error::new(
85		io::ErrorKind::Unsupported,
86		format!("cannot compress in {:?}", coding.token()),
87	)
88}
89
90/// Compress a streaming request body as its chunks arrive.
91///
92/// There is no compressed length to declare before the body ends, so the result goes out
93/// chunked. The encoder buffers on its own terms, so the bytes for one chunk the caller
94/// writes need not leave with it.
95// spec:ENC#what-a-compressed-request-sends
96pub fn compress_stream<S>(input: S, coding: Coding) -> io::Result<RequestStream>
97where
98	S: Stream<Item = io::Result<Bytes>> + Send + 'static,
99{
100	let reader = StreamReader::new(input);
101	Ok(match coding {
102		Coding::Gzip => encoder_stream(GzipEncoder::new(reader)),
103		Coding::Deflate => encoder_stream(ZlibEncoder::new(reader)),
104		Coding::Brotli => encoder_stream(BrotliEncoder::new(reader)),
105		Coding::Zstd => encoder_stream(ZstdEncoder::new(reader)),
106		other => return Err(unsupported(&other)),
107	})
108}
109
110fn encoder_stream<R>(reader: R) -> RequestStream
111where
112	R: tokio::io::AsyncRead + Send + 'static,
113{
114	Box::pin(ReaderStream::new(reader))
115}
web_faith_encoding/request.rs

web_faith_encoding/
request.rs
