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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
use memchr::memchr;
use tokio::io::{AsyncRead, AsyncReadExt};
use crate::util::allocate_vec;
const BUFFER_SIZE: usize = 32768;
pub struct LineReader {
buf: Box<[u8]>,
start_offset: usize,
end_offset: usize,
}
impl LineReader {
pub fn new() -> Self {
Self {
buf: allocate_vec(BUFFER_SIZE).into_boxed_slice(),
start_offset: 0usize,
end_offset: 0usize,
}
}
/// Create a LineReader pre-populated with initial data.
/// This is used when we've already buffered some data (e.g., from a failed TLS parse)
/// and need to parse it as HTTP. The data buffer is consumed directly without cloning.
/// If the data is smaller than BUFFER_SIZE, it will be extended to BUFFER_SIZE to allow
/// reading additional data from the stream. If larger, the data buffer itself is used
/// as-is (allowing larger initial buffers when needed).
pub fn new_with_data(mut data: Vec<u8>) -> Self {
let data_len = data.len();
if data_len < BUFFER_SIZE {
// Extend the buffer to BUFFER_SIZE to allow reading more data
data.resize(BUFFER_SIZE, 0);
Self {
buf: data.into_boxed_slice(),
start_offset: 0,
end_offset: data_len,
}
} else {
// Use the data buffer directly (larger than standard size)
Self {
buf: data.into_boxed_slice(),
start_offset: 0,
end_offset: data_len,
}
}
}
fn reset_buf_offset(&mut self) {
if self.start_offset == 0 {
return;
}
self.buf.copy_within(self.start_offset..self.end_offset, 0);
self.end_offset -= self.start_offset;
self.start_offset = 0;
}
pub async fn read_line_bytes<T>(&mut self, stream: &mut T) -> std::io::Result<&mut [u8]>
where
T: AsyncRead + Unpin,
{
let mut search_start_offset = self.start_offset;
loop {
let search_end_offset = self.end_offset;
match memchr(b'\n', &self.buf[search_start_offset..search_end_offset]) {
Some(pos) => {
let newline_pos = search_start_offset + pos;
if newline_pos == self.start_offset || self.buf[newline_pos - 1] != b'\r' {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"Line is not terminated by CRLF",
));
}
// strip crlf
let line = &mut self.buf[self.start_offset..newline_pos - 1];
let new_start_offset = newline_pos + 1;
if new_start_offset == search_end_offset {
self.start_offset = 0;
self.end_offset = 0;
} else {
self.start_offset = new_start_offset;
}
return Ok(line);
}
None => {
// There are no more newlines.
let previous_start_offset = self.start_offset;
self.read(stream).await?;
// Only search through new data.
if previous_start_offset != self.start_offset {
// this can only move to zero when reset_buf_offset is called.
assert!(self.start_offset == 0);
search_start_offset = search_end_offset - previous_start_offset;
} else {
search_start_offset = search_end_offset;
}
}
}
}
}
pub async fn read_line<T>(&mut self, stream: &mut T) -> std::io::Result<&str>
where
T: AsyncRead + Unpin,
{
let line_bytes = self.read_line_bytes(stream).await?;
std::str::from_utf8(line_bytes).map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Failed to decode utf8: {}", e),
)
})
}
pub fn unparsed_data(&self) -> &[u8] {
&self.buf[self.start_offset..self.end_offset]
}
pub fn into_buf(self) -> Box<[u8]> {
self.buf
}
async fn read<T>(&mut self, stream: &mut T) -> std::io::Result<()>
where
T: AsyncRead + Unpin,
{
// Note that read() needs to work for blocking I/O. So we need to return
// immediately after a single read() call.
if self.is_cache_full() {
return Err(std::io::Error::new(
std::io::ErrorKind::ConnectionAborted,
"cache is full, line is too long",
));
}
self.reset_buf_offset();
loop {
match stream.read(&mut self.buf[self.end_offset..]).await {
Ok(len) => {
if len == 0 {
// EOF
return Err(std::io::Error::new(
std::io::ErrorKind::ConnectionAborted,
"EOF while reading",
));
}
self.end_offset += len;
return Ok(());
}
Err(e) => {
if e.kind() == std::io::ErrorKind::Interrupted {
continue;
} else {
return Err(e);
}
}
}
}
}
fn is_cache_full(&self) -> bool {
self.start_offset == 0 && self.end_offset == self.buf.len()
}
}