ntex/http/
message.rs

1use std::{cell::Ref, cell::RefCell, cell::RefMut, fmt, net, rc::Rc};
2
3use bitflags::bitflags;
4use urly::Url;
5
6use crate::http::header::HeaderMap;
7use crate::http::{HeaderItem, Method, StatusCode, Version, h1::Codec};
8use crate::io::{IoBoxed, IoRef, types};
9use crate::util::Extensions;
10
11/// The connection behavior selected for an HTTP message.
12#[derive(Copy, Clone, PartialEq, Eq, Debug)]
13pub enum ConnectionType {
14    /// Close connection after response
15    Close,
16    /// Keep connection alive after response
17    KeepAlive,
18    /// Connection is upgraded to different type
19    Upgrade,
20}
21
22bitflags! {
23    #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
24    pub(crate) struct Flags: u8 {
25        const CLOSE       = 0b0000_0001;
26        const KEEP_ALIVE  = 0b0000_0010;
27        const UPGRADE     = 0b0000_0100;
28        const EXPECT      = 0b0000_1000;
29        const NO_CHUNKING = 0b0001_0000;
30    }
31}
32
33pub(crate) trait Head: Default + 'static + fmt::Debug {
34    fn clear(&mut self);
35
36    /// Calls `f` with the thread's pool, returns `None` if the pool is already destroyed.
37    fn with_pool<F, R>(f: F) -> Option<R>
38    where
39        F: FnOnce(&MessagePool<Self>) -> R;
40}
41
42#[derive(Clone, Debug)]
43pub(crate) enum CurrentIo {
44    Ref(IoRef),
45    Io(Rc<dyn IoAccess>),
46    None,
47}
48
49pub(crate) trait IoAccess: fmt::Debug {
50    fn get(&self) -> Option<&IoRef>;
51
52    fn take(&self) -> Option<(IoBoxed, Codec)>;
53}
54
55impl CurrentIo {
56    pub(crate) fn new(io: Rc<dyn IoAccess>) -> Self {
57        CurrentIo::Io(io)
58    }
59
60    pub(crate) fn as_ref(&self) -> Option<&IoRef> {
61        match self {
62            CurrentIo::Ref(io) => Some(io),
63            CurrentIo::Io(io) => io.get(),
64            CurrentIo::None => None,
65        }
66    }
67
68    pub(crate) fn take(&self) -> Option<(IoBoxed, Codec)> {
69        match self {
70            CurrentIo::Io(io) => io.take(),
71            _ => None,
72        }
73    }
74}
75
76/// The parsed metadata for an HTTP request.
77#[derive(Debug)]
78pub struct RequestHead {
79    /// Identifier of the connection that received the request.
80    pub id: usize,
81    /// Request URI.
82    pub uri: Url,
83    /// Request method.
84    pub method: Method,
85    /// HTTP protocol version.
86    pub version: Version,
87    /// Parsed request headers.
88    pub headers: HeaderMap,
89    /// Headers in their original order and with their original names.
90    ///
91    /// This collection is populated only when
92    /// [`HttpServiceConfig::set_headers_vec`](crate::http::HttpServiceConfig::set_headers_vec)
93    /// is enabled.
94    pub headers_vec: Vec<HeaderItem>,
95    /// Request-local type map.
96    pub extensions: RefCell<Extensions>,
97    pub(crate) io: CurrentIo,
98    pub(crate) flags: Flags,
99}
100
101impl Default for RequestHead {
102    fn default() -> RequestHead {
103        RequestHead {
104            id: 0,
105            io: CurrentIo::None,
106            uri: Url::new(),
107            method: Method::default(),
108            version: Version::HTTP_11,
109            headers: HeaderMap::with_capacity(16),
110            headers_vec: Vec::default(),
111            flags: Flags::empty(),
112            extensions: RefCell::new(Extensions::new()),
113        }
114    }
115}
116
117impl Head for RequestHead {
118    fn clear(&mut self) {
119        self.io = CurrentIo::None;
120        // the uri and headers may share the connection read buffer, a pooled
121        // head must not keep it alive
122        self.uri = Url::new();
123        self.flags = Flags::empty();
124        self.version = Version::HTTP_11;
125        self.headers.clear();
126        self.headers_vec.clear();
127        self.extensions.get_mut().clear();
128    }
129
130    fn with_pool<F, R>(f: F) -> Option<R>
131    where
132        F: FnOnce(&MessagePool<Self>) -> R,
133    {
134        REQUEST_POOL.try_with(|p| f(p)).ok()
135    }
136}
137
138impl RequestHead {
139    /// Returns the request extensions.
140    #[inline]
141    pub fn extensions(&self) -> Ref<'_, Extensions> {
142        self.extensions.borrow()
143    }
144
145    /// Returns mutable access to the request extensions.
146    #[inline]
147    pub fn extensions_mut(&self) -> RefMut<'_, Extensions> {
148        self.extensions.borrow_mut()
149    }
150
151    /// Returns the request headers.
152    pub fn headers(&self) -> &HeaderMap {
153        &self.headers
154    }
155
156    /// Returns mutable access to the request headers.
157    pub fn headers_mut(&mut self) -> &mut HeaderMap {
158        &mut self.headers
159    }
160
161    /// Returns headers preserved in their original order and casing.
162    pub fn headers_vec(&self) -> &[HeaderItem] {
163        &self.headers_vec
164    }
165
166    #[inline]
167    /// Sets the request connection behavior.
168    ///
169    /// Connection types are flags, and calling this method again does not clear
170    /// a previously set type. When several are set, `Close` takes precedence
171    /// over `KeepAlive`, which takes precedence over `Upgrade`.
172    pub fn set_connection_type(&mut self, ctype: ConnectionType) {
173        match ctype {
174            ConnectionType::Close => self.flags.insert(Flags::CLOSE),
175            ConnectionType::KeepAlive => self.flags.insert(Flags::KEEP_ALIVE),
176            ConnectionType::Upgrade => self.flags.insert(Flags::UPGRADE),
177        }
178    }
179
180    #[inline]
181    /// Returns the request connection behavior.
182    pub fn connection_type(&self) -> ConnectionType {
183        if self.flags.contains(Flags::CLOSE) {
184            ConnectionType::Close
185        } else if self.flags.contains(Flags::KEEP_ALIVE) {
186            ConnectionType::KeepAlive
187        } else if self.flags.contains(Flags::UPGRADE) {
188            ConnectionType::Upgrade
189        } else if self.version < Version::HTTP_11 {
190            ConnectionType::Close
191        } else {
192            ConnectionType::KeepAlive
193        }
194    }
195
196    #[inline]
197    /// Returns whether the request upgrades the connection.
198    pub fn upgrade(&self) -> bool {
199        self.flags.contains(Flags::UPGRADE)
200    }
201
202    #[inline]
203    /// Returns whether the request contains `Expect: 100-continue`.
204    pub fn expect(&self) -> bool {
205        self.flags.contains(Flags::EXPECT)
206    }
207
208    #[inline]
209    /// Returns whether chunked transfer encoding is allowed.
210    pub fn chunked(&self) -> bool {
211        !self.flags.contains(Flags::NO_CHUNKING)
212    }
213
214    #[inline]
215    /// Enables or disables chunked transfer encoding.
216    ///
217    /// A streaming request body without chunked coding requires a `Content-Length` header.
218    pub fn no_chunking(&mut self, val: bool) {
219        if val {
220            self.flags.insert(Flags::NO_CHUNKING);
221        } else {
222            self.flags.remove(Flags::NO_CHUNKING);
223        }
224    }
225
226    #[inline]
227    pub(crate) fn set_expect(&mut self) {
228        self.flags.insert(Flags::EXPECT);
229    }
230
231    #[inline]
232    pub(crate) fn set_upgrade(&mut self) {
233        self.flags.insert(Flags::UPGRADE);
234    }
235
236    /// Returns the peer socket address.
237    ///
238    /// Peer address is actual socket address, if proxy is used in front of
239    /// ntex http server, then peer address would be address of this proxy.
240    #[inline]
241    pub fn peer_addr(&self) -> Option<net::SocketAddr> {
242        self.io.as_ref().and_then(|io| {
243            io.query::<types::PeerAddr>()
244                .get()
245                .map(types::PeerAddr::into_inner)
246        })
247    }
248
249    /// Takes ownership of the I/O stream and HTTP/1 codec for an upgrade.
250    ///
251    /// The handle is installed only after an upgrade is acknowledged through
252    /// the HTTP/1 control service. This is a one-shot operation: subsequent
253    /// calls return [`None`].
254    pub fn take_io(&self) -> Option<(IoBoxed, Codec)> {
255        self.io.take()
256    }
257
258    #[doc(hidden)]
259    pub fn remove_io(&mut self) {
260        self.io = CurrentIo::None;
261    }
262}
263
264/// The metadata for an HTTP response.
265#[derive(Debug)]
266pub struct ResponseHead {
267    /// HTTP protocol version.
268    pub version: Version,
269    /// Response status code.
270    pub status: StatusCode,
271    /// Response headers.
272    pub headers: HeaderMap,
273    /// Headers in their original order and with their original names.
274    ///
275    /// This collection is populated when decoding a response with
276    /// [`HttpServiceConfig::set_headers_vec`](crate::http::HttpServiceConfig::set_headers_vec)
277    /// enabled.
278    pub headers_vec: Vec<HeaderItem>,
279    /// Custom reason phrase, or `None` to use the status code's standard phrase.
280    pub reason: Option<&'static str>,
281    pub(crate) io: CurrentIo,
282    pub(crate) extensions: RefCell<Extensions>,
283    flags: Flags,
284}
285
286impl ResponseHead {
287    /// Creates response metadata with the supplied status and HTTP version.
288    #[inline]
289    pub fn new(status: StatusCode, version: Version) -> ResponseHead {
290        ResponseHead {
291            status,
292            version,
293            headers: HeaderMap::with_capacity(12),
294            headers_vec: Vec::default(),
295            reason: None,
296            flags: Flags::empty(),
297            io: CurrentIo::None,
298            extensions: RefCell::new(Extensions::new()),
299        }
300    }
301
302    /// Returns the response extensions.
303    #[inline]
304    pub fn extensions(&self) -> Ref<'_, Extensions> {
305        self.extensions.borrow()
306    }
307
308    /// Returns mutable access to the response extensions.
309    #[inline]
310    pub fn extensions_mut(&self) -> RefMut<'_, Extensions> {
311        self.extensions.borrow_mut()
312    }
313
314    #[inline]
315    /// Returns the response headers.
316    pub fn headers(&self) -> &HeaderMap {
317        &self.headers
318    }
319
320    #[inline]
321    /// Returns mutable access to the response headers.
322    pub fn headers_mut(&mut self) -> &mut HeaderMap {
323        &mut self.headers
324    }
325
326    /// Returns headers preserved in their original order and casing.
327    pub fn headers_vec(&self) -> &[HeaderItem] {
328        &self.headers_vec
329    }
330
331    #[inline]
332    /// Sets the response connection behavior.
333    ///
334    /// Connection types are flags, and calling this method again does not clear
335    /// a previously set type. When several are set, `Close` takes precedence
336    /// over `KeepAlive`, which takes precedence over `Upgrade`.
337    pub fn set_connection_type(&mut self, ctype: ConnectionType) {
338        match ctype {
339            ConnectionType::Close => self.flags.insert(Flags::CLOSE),
340            ConnectionType::KeepAlive => self.flags.insert(Flags::KEEP_ALIVE),
341            ConnectionType::Upgrade => self.flags.insert(Flags::UPGRADE),
342        }
343    }
344
345    #[inline]
346    /// Returns the response's connection behavior.
347    pub fn connection_type(&self) -> ConnectionType {
348        if self.flags.contains(Flags::CLOSE) {
349            ConnectionType::Close
350        } else if self.flags.contains(Flags::KEEP_ALIVE) {
351            ConnectionType::KeepAlive
352        } else if self.flags.contains(Flags::UPGRADE) {
353            ConnectionType::Upgrade
354        } else if self.version < Version::HTTP_11 {
355            ConnectionType::Close
356        } else {
357            ConnectionType::KeepAlive
358        }
359    }
360
361    #[inline]
362    /// Returns whether the response keeps the connection open.
363    pub fn keep_alive(&self) -> bool {
364        self.connection_type() == ConnectionType::KeepAlive
365    }
366
367    #[inline]
368    /// Returns whether the response upgrades the connection.
369    pub fn upgrade(&self) -> bool {
370        self.connection_type() == ConnectionType::Upgrade
371    }
372
373    /// Returns the custom or canonical reason phrase.
374    ///
375    /// The reason phrase is empty for a status code without a canonical reason.
376    #[inline]
377    pub fn reason(&self) -> &str {
378        if let Some(reason) = self.reason {
379            reason
380        } else {
381            self.status.canonical_reason().unwrap_or("")
382        }
383    }
384
385    #[inline]
386    pub(crate) fn ctype(&self) -> Option<ConnectionType> {
387        if self.flags.contains(Flags::CLOSE) {
388            Some(ConnectionType::Close)
389        } else if self.flags.contains(Flags::KEEP_ALIVE) {
390            Some(ConnectionType::KeepAlive)
391        } else if self.flags.contains(Flags::UPGRADE) {
392            Some(ConnectionType::Upgrade)
393        } else {
394            None
395        }
396    }
397
398    #[inline]
399    /// Returns whether HTTP/1 chunked transfer encoding is allowed.
400    pub fn chunked(&self) -> bool {
401        !self.flags.contains(Flags::NO_CHUNKING)
402    }
403
404    #[inline]
405    /// Enables or disables HTTP/1 chunked transfer encoding.
406    pub fn no_chunking(&mut self, val: bool) {
407        if val {
408            self.flags.insert(Flags::NO_CHUNKING);
409        } else {
410            self.flags.remove(Flags::NO_CHUNKING);
411        }
412    }
413}
414
415impl Default for ResponseHead {
416    fn default() -> Self {
417        Self::new(StatusCode::default(), Version::default())
418    }
419}
420
421impl Head for ResponseHead {
422    fn clear(&mut self) {
423        self.reason = None;
424        self.headers.clear();
425        self.headers_vec.clear();
426        self.io = CurrentIo::None;
427        self.flags = Flags::empty();
428        self.extensions.get_mut().clear();
429    }
430
431    fn with_pool<F, R>(f: F) -> Option<R>
432    where
433        F: FnOnce(&MessagePool<Self>) -> R,
434    {
435        RESPONSE_POOL.try_with(|p| f(p)).ok()
436    }
437}
438
439#[derive(Debug)]
440pub(crate) struct Message<T: Head> {
441    head: Rc<T>,
442}
443
444impl<T: Head> Message<T> {
445    /// Get new message from the pool of objects
446    pub(crate) fn new() -> Self {
447        let head = T::with_pool(MessagePool::get_head)
448            .flatten()
449            .unwrap_or_else(|| Rc::new(T::default()));
450        Message { head }
451    }
452
453    /// Clears the head, so a cached message does not hold request data.
454    pub(crate) fn clear(&mut self) {
455        if let Some(head) = Rc::get_mut(&mut self.head) {
456            head.clear();
457        }
458    }
459}
460
461impl Message<ResponseHead> {
462    /// Get new message from the pool of objects
463    pub(crate) fn with_status(status: StatusCode) -> Self {
464        let mut msg = Self::new();
465        msg.status = status;
466        msg
467    }
468
469    /// Copies the response head into a new message, extensions are not copied.
470    pub(crate) fn copy(&self) -> Self {
471        let mut msg = Self::with_status(self.status);
472        msg.version = self.version;
473        msg.headers = self.headers.clone();
474        msg.headers_vec.clone_from(&self.headers_vec);
475        msg.reason = self.reason;
476        msg.io = self.io.clone();
477        msg.flags = self.flags;
478        msg
479    }
480}
481
482impl<T: Head> Clone for Message<T> {
483    fn clone(&self) -> Self {
484        Self {
485            head: self.head.clone(),
486        }
487    }
488}
489
490impl<T: Head> std::ops::Deref for Message<T> {
491    type Target = T;
492
493    fn deref(&self) -> &Self::Target {
494        self.head.as_ref()
495    }
496}
497
498impl<T: Head> std::ops::DerefMut for Message<T> {
499    fn deref_mut(&mut self) -> &mut Self::Target {
500        Rc::get_mut(&mut self.head).expect("Multiple copies exist")
501    }
502}
503
504impl<T: Head> Drop for Message<T> {
505    fn drop(&mut self) {
506        if let Some(head) = Rc::get_mut(&mut self.head) {
507            // clearing may drop other messages (e.g. stored in extensions),
508            // so the pool must not be borrowed at this point
509            head.clear();
510            T::with_pool(|pool| {
511                if let Ok(mut v) = pool.0.try_borrow_mut()
512                    && v.len() < 128
513                {
514                    v.push(self.head.clone());
515                }
516            });
517        }
518    }
519}
520
521/// Request's objects pool
522pub(crate) struct MessagePool<T: Head>(RefCell<Vec<Rc<T>>>);
523
524thread_local!(static REQUEST_POOL: MessagePool<RequestHead> = MessagePool::<RequestHead>::new());
525thread_local!(static RESPONSE_POOL: MessagePool<ResponseHead> = MessagePool::<ResponseHead>::new());
526
527impl<T: Head> MessagePool<T> {
528    fn new() -> MessagePool<T> {
529        MessagePool(RefCell::new(Vec::with_capacity(256)))
530    }
531
532    /// Get cleared head from the pool
533    #[inline]
534    fn get_head(&self) -> Option<Rc<T>> {
535        self.0.try_borrow_mut().ok()?.pop()
536    }
537}
538
539#[cfg(test)]
540mod tests {
541    use super::*;
542
543    #[test]
544    fn nested_message_drop() {
545        let inner = Message::<RequestHead>::new();
546        let outer = Message::<RequestHead>::new();
547        outer.extensions_mut().insert(inner);
548        let mut res = Message::<ResponseHead>::with_status(StatusCode::OK);
549        res.extensions
550            .get_mut()
551            .insert(Message::<ResponseHead>::new());
552        outer.extensions_mut().insert(res);
553        drop(outer);
554
555        let _ = Message::<RequestHead>::new();
556        let _ = Message::<ResponseHead>::new();
557    }
558
559    #[test]
560    fn message_drop_after_pool_destroyed() {
561        struct Holder(Option<Message<RequestHead>>, Option<Message<ResponseHead>>);
562
563        impl Drop for Holder {
564            fn drop(&mut self) {
565                drop(self.0.take());
566                drop(self.1.take());
567                drop(Message::<RequestHead>::new());
568            }
569        }
570
571        thread_local!(static HOLDER: RefCell<Option<Holder>> = const { RefCell::new(None) });
572
573        std::thread::spawn(|| {
574            // register holder before the pools, so it is destroyed after them
575            HOLDER.with(|_| ());
576            let req = Message::<RequestHead>::new();
577            let res = Message::<ResponseHead>::new();
578            HOLDER.with(|h| *h.borrow_mut() = Some(Holder(Some(req), Some(res))));
579        })
580        .join()
581        .unwrap();
582    }
583
584    #[test]
585    fn response_connection_type() {
586        let mut res = Message::<ResponseHead>::with_status(StatusCode::OK);
587        assert_eq!(res.connection_type(), ConnectionType::KeepAlive);
588        res.version = Version::HTTP_10;
589        assert_eq!(res.connection_type(), ConnectionType::Close);
590        res.set_connection_type(ConnectionType::KeepAlive);
591        assert_eq!(res.connection_type(), ConnectionType::KeepAlive);
592    }
593
594    #[test]
595    fn message_clone() {
596        let mut msg = Message::<RequestHead>::new();
597        msg.method = Method::POST;
598        let msg2 = msg.clone();
599        assert_eq!(msg2.method, Method::POST);
600        assert!(CurrentIo::None.take().is_none());
601    }
602}
ntex/http/message.rs

ntex/http/
message.rs
