ntex_io/
seal.rs

1use std::{any::Any, any::TypeId, fmt, io, ops, task::Context, task::Poll};
2
3use crate::{Filter, FilterCtx, Io, Readiness};
4
5/// Type-erased filter chain used by [`IoBoxed`].
6pub struct Sealed(pub(crate) Box<dyn Filter>);
7
8impl fmt::Debug for Sealed {
9    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
10        f.debug_struct("Sealed").finish()
11    }
12}
13
14impl Filter for Sealed {
15    #[inline]
16    fn query(&self, id: TypeId) -> Option<Box<dyn Any>> {
17        self.0.query(id)
18    }
19
20    #[inline]
21    fn process_read_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
22        self.0.process_read_buf(ctx)
23    }
24
25    #[inline]
26    fn process_write_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
27        self.0.process_write_buf(ctx)
28    }
29
30    #[inline]
31    fn shutdown(&self, ctx: &mut FilterCtx<'_>) -> io::Result<Poll<()>> {
32        self.0.shutdown(ctx)
33    }
34
35    #[inline]
36    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
37        self.0.poll_read_ready(cx)
38    }
39
40    #[inline]
41    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Readiness> {
42        self.0.poll_write_ready(cx)
43    }
44}
45
46#[derive(Debug)]
47/// An [`Io`] object whose filter-chain type has been erased.
48pub struct IoBoxed(Io<Sealed>);
49
50impl IoBoxed {
51    #[inline]
52    #[must_use]
53    /// Transfers the live I/O state into a new object.
54    ///
55    /// This does not clone the connection. The current object is replaced with
56    /// a stopped placeholder and should no longer be used for I/O.
57    pub fn take(&mut self) -> Self {
58        // SAFETY: `&mut self` rules out borrows of the inner `Io`
59        IoBoxed(unsafe { self.0.take() })
60    }
61}
62
63impl<F: Filter> From<Io<F>> for IoBoxed {
64    fn from(io: Io<F>) -> Self {
65        Self(io.seal())
66    }
67}
68
69impl ops::Deref for IoBoxed {
70    type Target = Io<Sealed>;
71
72    #[inline]
73    fn deref(&self) -> &Self::Target {
74        &self.0
75    }
76}
77
78impl From<IoBoxed> for Io<Sealed> {
79    fn from(value: IoBoxed) -> Self {
80        value.0
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use std::any::TypeId;
87
88    use ntex_bytes::Bytes;
89    use ntex_codec::BytesCodec;
90
91    use super::*;
92    use crate::{FilterBuf, FilterLayer, filter::NullFilter, testing::IoTest};
93
94    #[derive(Debug)]
95    struct Tagged;
96
97    impl FilterLayer for Tagged {
98        fn query(&self, id: TypeId) -> Option<Box<dyn Any>> {
99            (id == TypeId::of::<&'static str>()).then(|| Box::new("tagged") as Box<dyn Any>)
100        }
101
102        fn process_read_buf(&self, buf: &FilterBuf<'_>) -> io::Result<()> {
103            buf.with_read_buffers(|src, dst| {
104                if let Some(src) = src.take() {
105                    dst.extend_from_slice(&src);
106                }
107            });
108            Ok(())
109        }
110
111        fn process_write_buf(&self, buf: &FilterBuf<'_>) -> io::Result<()> {
112            buf.with_write_buffers(ntex_bytes::BytePages::move_to);
113            Ok(())
114        }
115    }
116
117    /// Filter chain wrapper built from the forwarding macros.
118    struct Wrapper<F> {
119        inner: F,
120    }
121
122    impl<F: Filter> Filter for Wrapper<F> {
123        crate::forward_ready!(inner);
124        crate::forward_query!(inner);
125        crate::forward_shutdown!(inner);
126
127        fn process_read_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
128            self.inner.process_read_buf(ctx)
129        }
130
131        fn process_write_buf(&self, ctx: &mut FilterCtx<'_>) -> io::Result<()> {
132            self.inner.process_write_buf(ctx)
133        }
134    }
135
136    #[ntex::test]
137    async fn sealed_chain_delegates_to_inner_filter() {
138        assert_eq!(format!("{:?}", Sealed(Box::new(NullFilter))), "Sealed");
139
140        let (client, server) = IoTest::create();
141        client.remote_buffer_cap(1024);
142
143        let io = Io::from(server).add_filter(Tagged).boxed();
144        let io: Io<Sealed> = io.into();
145        let io = io.map_filter(|inner| Wrapper { inner });
146        assert_eq!(io.query::<&'static str>().get(), Some("tagged"));
147        assert!(io.query::<u32>().get().is_none());
148
149        client.write("hello");
150        let item = io.recv(&BytesCodec).await.unwrap().unwrap();
151        assert_eq!(item, Bytes::from_static(b"hello"));
152
153        io.send(Bytes::from_static(b"world"), &BytesCodec)
154            .await
155            .unwrap();
156        assert_eq!(client.read().await.unwrap(), Bytes::from_static(b"world"));
157
158        io.shutdown().await.unwrap();
159        assert!(io.is_closed());
160    }
161}
ntex_io/seal.rs

ntex_io/
seal.rs
