rust_web_server/request/
body.rs

1//! The request body: bytes already read, or a stream the handler reads (#31).
2//!
3//! The server reads a request's body into memory before the handler runs,
4//! as it always has — a [`Body::Bytes`]. An application that wants to read a
5//! large upload as it arrives opts in per request
6//! ([`Application::streams_body`](crate::application::Application::streams_body))
7//! and gets a [`Body::Stream`] instead: the handler runs while the server is
8//! still receiving, and memory stays at a few chunks however big the body is.
9//!
10//! `Body` dereferences to `&[u8]`, so code that reads `request.body` as bytes
11//! keeps working; a streamed body derefs to an empty slice until
12//! [`Body::body_bytes`] reads it in.
13
14use std::fmt;
15use std::io::{self, Read};
16use std::sync::{Arc, Mutex};
17
18/// A request body.
19pub enum Body {
20    /// The whole body, already read.
21    Bytes(Vec<u8>),
22    /// A body still arriving; read it with [`Read`] (`(&mut body).read(..)`,
23    /// `std::io::copy`), [`Body::into_reader`] or [`Body::body_bytes`].
24    /// Clones share the one stream: what one reads, the others don't see.
25    Stream(Arc<Mutex<Box<dyn Read + Send>>>),
26}
27
28impl Body {
29    /// An empty body.
30    pub fn empty() -> Self {
31        Body::Bytes(Vec::new())
32    }
33
34    /// A body streamed from `reader`.
35    pub fn stream(reader: impl Read + Send + 'static) -> Self {
36        Body::Stream(Arc::new(Mutex::new(Box::new(reader))))
37    }
38
39    /// `true` for a body still arriving.
40    pub fn is_stream(&self) -> bool {
41        matches!(self, Body::Stream(_))
42    }
43
44    /// The bytes read so far: the whole body, or empty for a stream.
45    pub fn as_bytes(&self) -> &[u8] {
46        match self {
47            Body::Bytes(bytes) => bytes,
48            Body::Stream(_) => &[],
49        }
50    }
51
52    /// The whole body: a stream is read to the end first (and stays read —
53    /// the body is `Bytes` afterwards). Fails if reading the stream fails,
54    /// e.g. the client went away or the body passed the size limit.
55    pub fn body_bytes(&mut self) -> io::Result<&[u8]> {
56        if let Body::Stream(stream) = self {
57            let mut bytes = Vec::new();
58            stream.lock().map_err(|_| io::Error::other("body stream poisoned"))?.read_to_end(&mut bytes)?;
59            *self = Body::Bytes(bytes);
60        }
61        Ok(self.as_bytes())
62    }
63
64    /// The body as one reader, whatever it is.
65    pub fn into_reader(self) -> Box<dyn Read + Send> {
66        match self {
67            Body::Bytes(bytes) => Box::new(io::Cursor::new(bytes)),
68            Body::Stream(stream) => Box::new(SharedReader(stream)),
69        }
70    }
71
72    /// The whole body as a `Vec<u8>` (a stream is read to the end).
73    pub fn into_vec(mut self) -> io::Result<Vec<u8>> {
74        self.body_bytes()?;
75        match self {
76            Body::Bytes(bytes) => Ok(bytes),
77            Body::Stream(_) => unreachable!("body_bytes turns a stream into bytes"),
78        }
79    }
80
81    /// Mutable access to the bytes (a stream is read in first).
82    pub fn bytes_mut(&mut self) -> io::Result<&mut Vec<u8>> {
83        self.body_bytes()?;
84        match self {
85            Body::Bytes(bytes) => Ok(bytes),
86            Body::Stream(_) => unreachable!("body_bytes turns a stream into bytes"),
87        }
88    }
89}
90
91struct SharedReader(Arc<Mutex<Box<dyn Read + Send>>>);
92
93impl Read for SharedReader {
94    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
95        self.0.lock().map_err(|_| io::Error::other("body stream poisoned"))?.read(buf)
96    }
97}
98
99/// Reading a `Bytes` body yields nothing (its bytes are in the slice);
100/// reading a stream yields the next bytes received.
101impl Read for &Body {
102    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
103        match self {
104            Body::Bytes(_) => Ok(0),
105            Body::Stream(stream) => stream.lock().map_err(|_| io::Error::other("body stream poisoned"))?.read(buf),
106        }
107    }
108}
109
110impl Default for Body {
111    fn default() -> Self {
112        Body::empty()
113    }
114}
115
116impl std::ops::Deref for Body {
117    type Target = [u8];
118    fn deref(&self) -> &[u8] {
119        self.as_bytes()
120    }
121}
122
123impl AsRef<[u8]> for Body {
124    fn as_ref(&self) -> &[u8] {
125        self.as_bytes()
126    }
127}
128
129impl Clone for Body {
130    fn clone(&self) -> Self {
131        match self {
132            Body::Bytes(bytes) => Body::Bytes(bytes.clone()),
133            Body::Stream(stream) => Body::Stream(Arc::clone(stream)),
134        }
135    }
136}
137
138/// Bytes compare by content; a stream equals only its own clones.
139impl PartialEq for Body {
140    fn eq(&self, other: &Self) -> bool {
141        match (self, other) {
142            (Body::Bytes(a), Body::Bytes(b)) => a == b,
143            (Body::Stream(a), Body::Stream(b)) => Arc::ptr_eq(a, b),
144            _ => false,
145        }
146    }
147}
148
149impl Eq for Body {}
150
151impl PartialEq<Vec<u8>> for Body {
152    fn eq(&self, other: &Vec<u8>) -> bool {
153        matches!(self, Body::Bytes(bytes) if bytes == other)
154    }
155}
156
157impl PartialEq<[u8]> for Body {
158    fn eq(&self, other: &[u8]) -> bool {
159        matches!(self, Body::Bytes(bytes) if bytes.as_slice() == other)
160    }
161}
162
163impl PartialEq<Body> for Vec<u8> {
164    fn eq(&self, other: &Body) -> bool {
165        other == self
166    }
167}
168
169impl PartialEq<Body> for &[u8] {
170    fn eq(&self, other: &Body) -> bool {
171        other == *self
172    }
173}
174
175impl PartialEq<Body> for [u8] {
176    fn eq(&self, other: &Body) -> bool {
177        other == self
178    }
179}
180
181impl fmt::Debug for Body {
182    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
183        match self {
184            Body::Bytes(bytes) => f.debug_tuple("Bytes").field(&bytes.len()).finish(),
185            Body::Stream(_) => f.write_str("Stream"),
186        }
187    }
188}
189
190impl From<Vec<u8>> for Body {
191    fn from(bytes: Vec<u8>) -> Self {
192        Body::Bytes(bytes)
193    }
194}
195
196impl From<&[u8]> for Body {
197    fn from(bytes: &[u8]) -> Self {
198        Body::Bytes(bytes.to_vec())
199    }
200}
201
202impl<const N: usize> From<&[u8; N]> for Body {
203    fn from(bytes: &[u8; N]) -> Self {
204        Body::Bytes(bytes.to_vec())
205    }
206}
207
208impl From<String> for Body {
209    fn from(text: String) -> Self {
210        Body::Bytes(text.into_bytes())
211    }
212}
213
214impl From<&str> for Body {
215    fn from(text: &str) -> Self {
216        Body::Bytes(text.as_bytes().to_vec())
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use super::Body;
223    use std::io::Read;
224
225    #[test]
226    fn bytes_deref_compare_and_convert() {
227        let body: Body = b"abc".to_vec().into();
228        assert_eq!(3, body.len());
229        assert_eq!(b"abc", &body[..]);
230        assert_eq!(b"abc".to_vec(), body);
231        assert!(body == b"abc".to_vec());
232        assert_eq!(Body::from("abc"), Body::from(&b"abc"[..]));
233        assert!(!body.is_stream());
234        assert_eq!(b"abc".to_vec(), body.clone().into_vec().unwrap());
235        assert!(Body::default().is_empty());
236    }
237
238    #[test]
239    fn a_stream_reads_once_and_becomes_bytes() {
240        let mut body = Body::stream(std::io::Cursor::new(b"streamed".to_vec()));
241        assert!(body.is_stream());
242        assert!(body.is_empty(), "a stream derefs to nothing until read");
243        assert_eq!(b"streamed", body.body_bytes().unwrap());
244        assert!(!body.is_stream());
245        assert_eq!(b"streamed", &body[..]);
246    }
247
248    #[test]
249    fn clones_of_a_stream_share_it() {
250        let body = Body::stream(std::io::Cursor::new(b"0123456789".to_vec()));
251        let other = body.clone();
252        assert_eq!(body, other);
253        let mut first = [0u8; 4];
254        (&body).read_exact(&mut first).unwrap();
255        let mut rest = Vec::new();
256        other.into_reader().read_to_end(&mut rest).unwrap();
257        assert_eq!(b"0123", &first);
258        assert_eq!(b"456789".to_vec(), rest);
259        assert_ne!(Body::stream(std::io::empty()), Body::stream(std::io::empty()));
260    }
261
262    #[test]
263    fn a_failing_stream_is_an_error() {
264        struct Broken;
265        impl Read for Broken {
266            fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
267                Err(std::io::Error::other("client went away"))
268            }
269        }
270        let mut body = Body::stream(Broken);
271        assert!(body.body_bytes().is_err());
272    }
273}
rust_web_server/request/body.rs

rust_web_server/request/
body.rs
