ntex_io/
filter.rs

1use std::{any, cell::Cell, io, task::Context, task::Poll};
2
3use crate::{FilterCtx, FilterLayer, IoRef, Readiness};
4
5#[derive(Debug)]
6/// Default `Io` filter
7pub struct Base(IoRef);
8
9impl Base {
10    pub(crate) fn new(inner: IoRef) -> Self {
11        Base(inner)
12    }
13}
14
15#[derive(Debug)]
16pub struct Layer<F, L = Base>(pub(crate) F, L, Cell<bool>);
17
18impl<F: FilterLayer, L: Filter> Layer<F, L> {
19    pub(crate) fn new(f: F, l: L) -> Self {
20        Self(f, l, Cell::new(false))
21    }
22}
23
24pub(crate) struct NullFilter;
25
26const NULL: NullFilter = NullFilter;
27
28impl NullFilter {
29    pub(super) const fn get() -> &'static dyn Filter {
30        &NULL
31    }
32}
33
34pub trait Filter: 'static {
35    /// Accesses internal filter information.
36    fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>>;
37
38    /// Processes incoming read-buffer data.
39    fn process_read_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()>;
40
41    /// Processes outgoing write-buffer data.
42    fn process_write_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()>;
43
44    /// Performs a graceful shutdown of the filter.
45    fn shutdown(&self, ctx: &mut FilterCtx<'_>) -> io::Result<Poll<()>>;
46
47    /// Checks whether read operations may proceed.
48    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness>;
49
50    /// Checks whether write operations may proceed.
51    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness>;
52}
53
54impl Filter for Base {
55    fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
56        if let Some(hnd) = self.0.0.handle.take() {
57            let res = hnd.query(id);
58            self.0.0.handle.set(Some(hnd));
59            res
60        } else {
61            None
62        }
63    }
64
65    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
66        let st = &self.0.0;
67        if st.flags.is_closed() {
68            Poll::Ready(Readiness::Terminate)
69        } else {
70            st.read_task.register(cx.waker());
71
72            if st.flags.is_stopping_filters() {
73                Poll::Ready(Readiness::Ready)
74            } else if st.flags.is_read_paused_or_backpressure() {
75                // read buffer is fulled of is not processed by dispatcher yet
76                Poll::Pending
77            } else {
78                Poll::Ready(Readiness::Ready)
79            }
80        }
81    }
82
83    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
84        if self.0.0.flags.is_closed() {
85            Poll::Ready(Readiness::Terminate)
86        } else {
87            self.0.0.write_task.register(cx.waker());
88
89            if self.0.0.flags.is_stopping() {
90                Poll::Ready(Readiness::Shutdown)
91            } else if self.0.0.flags.is_write_paused() {
92                Poll::Pending
93            } else {
94                Poll::Ready(Readiness::Ready)
95            }
96        }
97    }
98
99    #[inline]
100    fn process_read_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
101        Ok(())
102    }
103
104    #[inline]
105    fn process_write_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
106        Ok(())
107    }
108
109    #[inline]
110    fn shutdown(&self, _: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
111        Ok(Poll::Ready(()))
112    }
113}
114
115impl<F, L> Filter for Layer<F, L>
116where
117    F: FilterLayer,
118    L: Filter,
119{
120    #[inline]
121    fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
122        self.0.query(id).or_else(|| self.1.query(id))
123    }
124
125    #[inline]
126    fn shutdown(&self, ctx: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
127        if !self.2.get() {
128            if ctx.with_buffer(|buf| self.0.shutdown(buf))?.is_ready() {
129                self.process_write_buf(ctx)?;
130                self.2.set(true);
131
132                // Discard the write buffer; it won't be processed
133                ctx.clear_write_buf();
134            } else {
135                return Ok(Poll::Pending);
136            }
137        }
138        ctx.with_next(|ctx| self.1.shutdown(ctx))
139    }
140
141    #[inline]
142    fn process_read_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
143        ctx.with_next(|ctx| self.1.process_read_buf(ctx))?;
144        if self.2.get() {
145            Ok(())
146        } else {
147            ctx.with_buffer(|buf| self.0.process_read_buf(buf))
148        }
149    }
150
151    #[inline]
152    fn process_write_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
153        if !self.2.get() {
154            ctx.with_buffer(|buf| self.0.process_write_buf(buf))?;
155        }
156        ctx.with_next(|ctx| self.1.process_write_buf(ctx))
157    }
158
159    #[inline]
160    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
161        self.1.poll_read_ready(cx)
162    }
163
164    #[inline]
165    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
166        self.1.poll_write_ready(cx)
167    }
168}
169
170impl Filter for NullFilter {
171    #[inline]
172    fn query(&self, _: any::TypeId) -> Option<Box<dyn any::Any>> {
173        None
174    }
175
176    #[inline]
177    fn poll_read_ready(&self, _: &mut Context<'_>) -> Poll<Readiness> {
178        Poll::Ready(Readiness::Terminate)
179    }
180
181    #[inline]
182    fn poll_write_ready(&self, _: &mut Context<'_>) -> Poll<Readiness> {
183        Poll::Ready(Readiness::Terminate)
184    }
185
186    #[inline]
187    fn process_read_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
188        Ok(())
189    }
190
191    #[inline]
192    fn process_write_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
193        Ok(())
194    }
195
196    #[inline]
197    fn shutdown(&self, _: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
198        Ok(Poll::Ready(()))
199    }
200}
ntex_io/filter.rs

ntex_io/
filter.rs
