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    #[inline]
66    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
67        let st = &self.0.0;
68        if st.flags.is_closed() {
69            Poll::Ready(Readiness::Terminate)
70        } else {
71            st.read_task.register(cx.waker());
72
73            if st.flags.is_stopping_filters() {
74                Poll::Ready(Readiness::Ready)
75            } else if st.flags.is_read_paused_or_backpressure() {
76                // read buffer is fulled of is not processed by dispatcher yet
77                Poll::Pending
78            } else {
79                Poll::Ready(Readiness::Ready)
80            }
81        }
82    }
83
84    #[inline]
85    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
86        if self.0.0.flags.is_closed() {
87            Poll::Ready(Readiness::Terminate)
88        } else {
89            self.0.0.write_task.register(cx.waker());
90
91            if self.0.0.flags.is_stopping() {
92                Poll::Ready(Readiness::Shutdown)
93            } else if self.0.0.flags.is_write_paused() {
94                Poll::Pending
95            } else {
96                Poll::Ready(Readiness::Ready)
97            }
98        }
99    }
100
101    #[inline]
102    fn process_read_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
103        Ok(())
104    }
105
106    #[inline]
107    fn process_write_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
108        Ok(())
109    }
110
111    #[inline]
112    fn shutdown(&self, _: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
113        Ok(Poll::Ready(()))
114    }
115}
116
117impl<F, L> Filter for Layer<F, L>
118where
119    F: FilterLayer,
120    L: Filter,
121{
122    #[inline]
123    fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
124        self.0.query(id).or_else(|| self.1.query(id))
125    }
126
127    #[inline]
128    fn shutdown(&self, ctx: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
129        if !self.2.get() {
130            if ctx.with_buffer(|buf| self.0.shutdown(buf))?.is_ready() {
131                self.process_write_buf(ctx)?;
132                self.2.set(true);
133
134                // Discard the write buffer; it won't be processed
135                ctx.clear_write_buf();
136            } else {
137                return Ok(Poll::Pending);
138            }
139        }
140        ctx.with_next(|ctx| self.1.shutdown(ctx))
141    }
142
143    #[inline]
144    fn process_read_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
145        ctx.with_next(|ctx| self.1.process_read_buf(ctx))?;
146        if self.2.get() {
147            Ok(())
148        } else {
149            ctx.with_buffer(|buf| self.0.process_read_buf(buf))
150        }
151    }
152
153    #[inline]
154    fn process_write_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
155        if !self.2.get() {
156            ctx.with_buffer(|buf| self.0.process_write_buf(buf))?;
157        }
158        ctx.with_next(|ctx| self.1.process_write_buf(ctx))
159    }
160
161    #[inline]
162    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
163        self.1.poll_read_ready(cx)
164    }
165
166    #[inline]
167    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
168        self.1.poll_write_ready(cx)
169    }
170}
171
172impl Filter for NullFilter {
173    #[inline]
174    fn query(&self, _: any::TypeId) -> Option<Box<dyn any::Any>> {
175        None
176    }
177
178    #[inline]
179    fn poll_read_ready(&self, _: &mut Context<'_>) -> Poll<Readiness> {
180        Poll::Ready(Readiness::Terminate)
181    }
182
183    #[inline]
184    fn poll_write_ready(&self, _: &mut Context<'_>) -> Poll<Readiness> {
185        Poll::Ready(Readiness::Terminate)
186    }
187
188    #[inline]
189    fn process_read_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
190        Ok(())
191    }
192
193    #[inline]
194    fn process_write_buf(&self, _: &mut FilterCtx<'_>) -> io::Result<()> {
195        Ok(())
196    }
197
198    #[inline]
199    fn shutdown(&self, _: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
200        Ok(Poll::Ready(()))
201    }
202}
ntex_io/filter.rs

ntex_io/
filter.rs
