warp/filters/
multipart.rs

1//! Multipart body filters
2//!
3//! [`Filter`](crate::Filter)s that extract a multipart body for a route.
4
5use std::error::Error as StdError;
6use std::fmt::{Display, Formatter};
7use std::future::Future;
8use std::pin::Pin;
9use std::task::{Context, Poll};
10use std::{fmt, io};
11
12use bytes::{Buf, Bytes};
13use futures_util::{future, Stream};
14use headers::ContentType;
15use hyper::Body;
16use mime::Mime;
17use multer::{Field as PartInner, Multipart as FormDataInner};
18
19use crate::filter::{Filter, FilterBase, Internal};
20use crate::reject::{self, Rejection};
21
22// If not otherwise configured, default to 2MB.
23const DEFAULT_FORM_DATA_MAX_LENGTH: u64 = 1024 * 1024 * 2;
24
25/// A [`Filter`](crate::Filter) to extract a `multipart/form-data` body from a request.
26///
27/// Create with the `warp::multipart::form()` function.
28#[derive(Debug, Clone)]
29pub struct FormOptions {
30    max_length: Option<u64>,
31}
32
33/// A `Stream` of multipart/form-data `Part`s.
34///
35/// Extracted with a `warp::multipart::form` filter.
36pub struct FormData {
37    inner: FormDataInner<'static>,
38}
39
40/// A single "part" of a multipart/form-data body.
41///
42/// Yielded from the `FormData` stream.
43pub struct Part {
44    part: PartInner<'static>,
45}
46
47/// Create a [`Filter`](crate::Filter) to extract a `multipart/form-data` body from a request.
48///
49/// The extracted `FormData` type is a `Stream` of `Part`s, and each `Part`
50/// in turn is a `Stream` of bytes.
51pub fn form() -> FormOptions {
52    FormOptions {
53        max_length: Some(DEFAULT_FORM_DATA_MAX_LENGTH),
54    }
55}
56
57// ===== impl Form =====
58
59impl FormOptions {
60    /// Set the maximum byte length allowed for this body.
61    ///
62    /// `max_length(None)` means that maximum byte length is not checked.
63    /// Defaults to 2MB.
64    pub fn max_length(mut self, max: impl Into<Option<u64>>) -> Self {
65        self.max_length = max.into();
66        self
67    }
68}
69
70type FormFut = Pin<Box<dyn Future<Output = Result<(FormData,), Rejection>> + Send>>;
71
72impl FilterBase for FormOptions {
73    type Extract = (FormData,);
74    type Error = Rejection;
75    type Future = FormFut;
76
77    fn filter(&self, _: Internal) -> Self::Future {
78        let boundary = super::header::header2::<ContentType>().and_then(|ct| {
79            let mime = Mime::from(ct);
80            let mime = mime
81                .get_param("boundary")
82                .map(|v| v.to_string())
83                .ok_or_else(|| reject::invalid_header("content-type"));
84            future::ready(mime)
85        });
86
87        let filt = boundary
88            .and(super::body::body())
89            .map(|boundary: String, body| {
90                let body = BodyIoError(body);
91                FormData {
92                    inner: FormDataInner::new(body, &boundary),
93                }
94            });
95
96        if let Some(max_length) = self.max_length {
97            Box::pin(
98                super::body::content_length_limit(max_length)
99                    .and(filt)
100                    .filter(Internal),
101            )
102        } else {
103            Box::pin(filt.filter(Internal))
104        }
105    }
106}
107
108// ===== impl FormData =====
109
110impl fmt::Debug for FormData {
111    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
112        f.debug_struct("FormData").finish()
113    }
114}
115
116impl Stream for FormData {
117    type Item = Result<Part, crate::Error>;
118
119    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
120        match self.inner.poll_next_field(cx) {
121            Poll::Pending => Poll::Pending,
122            Poll::Ready(Ok(Some(part))) => {
123                if part.name().is_some() || part.file_name().is_some() {
124                    Poll::Ready(Some(Ok(Part { part })))
125                } else {
126                    Poll::Ready(Some(Err(crate::Error::new(MultipartFieldMissingName))))
127                }
128            }
129            Poll::Ready(Ok(None)) => Poll::Ready(None),
130            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(crate::Error::new(err)))),
131        }
132    }
133}
134
135// ===== impl Part =====
136
137impl Part {
138    /// Get the name of this part.
139    pub fn name(&self) -> &str {
140        self.part
141            .name()
142            .unwrap_or_else(|| self.part.file_name().expect("checked for name previously"))
143    }
144
145    /// Get the filename of this part, if present.
146    pub fn filename(&self) -> Option<&str> {
147        self.part.file_name()
148    }
149
150    /// Get the content-type of this part, if present.
151    pub fn content_type(&self) -> Option<&str> {
152        let content_type = self.part.content_type();
153        content_type.map(|t| t.as_ref())
154    }
155
156    /// Asynchronously get some of the data for this `Part`.
157    pub async fn data(&mut self) -> Option<Result<impl Buf, crate::Error>> {
158        future::poll_fn(|cx| self.poll_next(cx)).await
159    }
160
161    /// Convert this `Part` into a `Stream` of `Buf`s.
162    pub fn stream(self) -> impl Stream<Item = Result<impl Buf, crate::Error>> {
163        PartStream(self)
164    }
165
166    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, crate::Error>>> {
167        match Pin::new(&mut self.part).poll_next(cx) {
168            Poll::Pending => Poll::Pending,
169            Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(bytes))),
170            Poll::Ready(None) => Poll::Ready(None),
171            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(crate::Error::new(err)))),
172        }
173    }
174}
175
176impl fmt::Debug for Part {
177    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
178        let mut builder = f.debug_struct("Part");
179        builder.field("name", &self.name());
180
181        if let Some(ref filename) = self.part.file_name() {
182            builder.field("filename", filename);
183        }
184
185        if let Some(ref mime) = self.part.content_type() {
186            builder.field("content_type", mime);
187        }
188
189        builder.finish()
190    }
191}
192
193struct PartStream(Part);
194
195impl Stream for PartStream {
196    type Item = Result<Bytes, crate::Error>;
197
198    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
199        self.0.poll_next(cx)
200    }
201}
202
203struct BodyIoError(Body);
204
205impl Stream for BodyIoError {
206    type Item = io::Result<Bytes>;
207
208    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
209        match Pin::new(&mut self.0).poll_next(cx) {
210            Poll::Pending => Poll::Pending,
211            Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(bytes))),
212            Poll::Ready(None) => Poll::Ready(None),
213            Poll::Ready(Some(Err(err))) => {
214                Poll::Ready(Some(Err(io::Error::new(io::ErrorKind::Other, err))))
215            }
216        }
217    }
218}
219
220/// An error used when a multipart field is missing a name.
221#[derive(Debug)]
222struct MultipartFieldMissingName;
223
224impl Display for MultipartFieldMissingName {
225    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
226        write!(f, "Multipart field is missing a name")
227    }
228}
229
230impl StdError for MultipartFieldMissingName {}
warp/filters/multipart.rs

warp/filters/
multipart.rs
