1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
//! Reading a message head off a stream, under the bound the edge enforces.
//!
//! Split from [`crate::exchange`] because it answers a different question.
//! The edge decides what a head *means* — how the body is framed, whether
//! the credential admits, whether the backend is contacted at all. This
//! decides only when there is a head to look at, and refuses one that is
//! too large to be worth buffering.
//!
//! Generic over [`AsyncRead`], so nothing here learns where the bytes came
//! from. The request and response readers were written twice and were the
//! same loop both times; they are one loop now, and the caller supplies the
//! parser.
use tokio::io::{AsyncRead, AsyncReadExt};
use crate::http_head::{self, HeadError, MAX_HEAD_BYTES, RequestHead, ResponseHead};
/// One pull from the stream. Small on purpose: a head is a few hundred
/// bytes and this is re-parsed from the start on every fill.
const PULL: usize = 4096;
/// What a head reader hands back: the head, and the bytes that arrived
/// after it. Named because the two readers and the loop behind them all
/// spell it, and spelling it three times is what the lint is objecting to.
pub(crate) type Read<H> = std::io::Result<Result<(H, Vec<u8>), HeadError>>;
/// A head parser, as [`read`] consumes one. `Ok(None)` means "a valid
/// prefix, send more".
type Parse<H> = fn(&[u8]) -> Result<Option<(H, usize)>, HeadError>;
/// Read until the request head is complete, or the bound is reached.
///
/// `prefix` is bytes already taken off the stream that belong to this head.
/// The returned `Vec` is the tail — whatever arrived after the head, which
/// is usually the start of the body.
///
/// The outer `Result` is transport failure; the inner one is a head this
/// edge refuses. They are different things: one means the stream broke, the
/// other means the peer sent something it should not have.
pub(crate) async fn request<S: AsyncRead + Unpin>(
stream: &mut S,
prefix: Vec<u8>,
) -> Read<RequestHead> {
read(stream, prefix, http_head::parse_request).await
}
/// The response twin of [`request`].
///
/// `prefix` earns its place here rather than being an artefact of sharing
/// the loop: an interim (`1xx`) response is a complete head followed
/// immediately by another, and the bytes that arrived with the interim one
/// are the start of the head after it.
pub(crate) async fn response<S: AsyncRead + Unpin>(
stream: &mut S,
prefix: Vec<u8>,
) -> Read<ResponseHead> {
read(stream, prefix, http_head::parse_response).await
}
/// The loop both readers are.
async fn read<S, H>(stream: &mut S, prefix: Vec<u8>, parse: Parse<H>) -> Read<H>
where
S: AsyncRead + Unpin,
{
let mut buf = prefix;
loop {
match parse(&buf) {
Ok(Some((head, consumed))) => return Ok(Ok((head, buf[consumed..].to_vec()))),
Ok(None) => {}
Err(e) => return Ok(Err(e)),
}
if buf.len() > MAX_HEAD_BYTES {
return Ok(Err(HeadError::TooLarge));
}
let mut next = [0u8; PULL];
let n = stream.read(&mut next).await?;
if n == 0 {
// The peer stopped mid-head. Nothing to answer, and nothing to
// forward.
return Ok(Err(HeadError::Malformed));
}
buf.extend_from_slice(&next[..n]);
}
}