ntex_io/
lib.rs

1//! Utilities for abstructing io streams
2#![deny(clippy::pedantic)]
3#![allow(
4    clippy::missing_fields_in_debug,
5    clippy::missing_errors_doc,
6    clippy::missing_panics_doc,
7    clippy::must_use_candidate
8)]
9use std::io::{Error as IoError, Result as IoResult};
10use std::{any::Any, any::TypeId, fmt, task::Poll};
11
12pub mod cfg;
13pub mod testing;
14pub mod types;
15
16mod buf;
17mod ctx;
18mod filter;
19mod filterptr;
20mod flags;
21mod framed;
22mod io;
23mod ioref;
24mod macros;
25mod ops;
26mod seal;
27mod utils;
28
29use ntex_codec::Decoder;
30
31pub use self::buf::{FilterBuf, FilterCtx};
32pub use self::cfg::IoConfig;
33pub use self::ctx::IoContext;
34pub use self::filter::{Base, Filter, Layer};
35pub use self::framed::Framed;
36pub use self::io::{Io, IoRef, OnDisconnect};
37pub use self::ops::{Id, TimerHandle};
38pub use self::seal::{IoBoxed, Sealed};
39pub use self::utils::{Decoded, seal};
40
41#[doc(hidden)]
42pub use self::flags::Flags;
43
44/// Filter readiness state.
45#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
46pub enum Readiness {
47    /// The I/O task may proceed with I/O operations.
48    Ready,
49    /// Initiates a graceful I/O shutdown.
50    Shutdown,
51    /// Immediately terminates the I/O stream.
52    Terminate,
53}
54
55impl Readiness {
56    /// Merges two readiness states.
57    pub fn merge(val1: Poll<Readiness>, val2: Poll<Readiness>) -> Poll<Readiness> {
58        match val1 {
59            Poll::Pending => Poll::Pending,
60            Poll::Ready(Readiness::Ready) => val2,
61            Poll::Ready(Readiness::Terminate) => Poll::Ready(Readiness::Terminate),
62            Poll::Ready(Readiness::Shutdown) => {
63                if val2 == Poll::Ready(Readiness::Terminate) {
64                    Poll::Ready(Readiness::Terminate)
65                } else {
66                    Poll::Ready(Readiness::Shutdown)
67                }
68            }
69        }
70    }
71}
72
73#[allow(unused_variables)]
74pub trait FilterLayer: fmt::Debug + 'static {
75    /// Accesses internal filter information.
76    fn query(&self, id: TypeId) -> Option<Box<dyn Any>> {
77        None
78    }
79
80    /// Processes incoming read-buffer data.
81    fn process_read_buf(&self, buf: &FilterBuf<'_>) -> IoResult<()>;
82
83    /// Processes outgoing write-buffer data.
84    fn process_write_buf(&self, buf: &FilterBuf<'_>) -> IoResult<()>;
85
86    /// Performs a graceful shutdown of the filter.
87    fn shutdown(&self, buf: &FilterBuf<'_>) -> IoResult<Poll<()>> {
88        Ok(Poll::Ready(()))
89    }
90}
91
92pub trait IoStream {
93    fn start(self, _: IoContext) -> Box<dyn Handle>;
94}
95
96#[doc(hidden)]
97/// Callbacks for filter processing
98pub trait IoCallbacks {
99    /// Get called before processing read or write buffers chain
100    fn before_processing(&self, io: &IoRef);
101
102    /// Get called after processing read or write buffers chain
103    fn after_processing(&self, io: &IoRef);
104}
105
106pub trait Handle {
107    fn query(&self, _: TypeId) -> Option<Box<dyn Any>> {
108        None
109    }
110
111    #[inline]
112    /// Initiate io write operation
113    fn write(&self, _: &IoContext) {}
114
115    #[inline]
116    /// Called when readiness changes
117    fn notify(&self, ctx: &IoContext) {
118        ctx.notify();
119    }
120}
121
122/// Current status of the I/O state.
123#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
124pub enum IoTaskStatus {
125    /// Continue performing I/O operations.
126    Io,
127    /// Pause I/O processing temporarily.
128    Pause,
129    /// Stop the I/O task.
130    Stop,
131}
132
133/// I/O status update events.
134#[derive(Debug)]
135pub enum IoStatusUpdate {
136    /// Keep-alive timeout has occurred.
137    KeepAlive,
138    /// Write backpressure is currently active.
139    WriteBackpressure,
140    /// Peer has disconnected.
141    PeerGone(Option<IoError>),
142}
143
144/// Errors that can occur while receiving data.
145pub enum RecvError<U: Decoder> {
146    /// A keep-alive timeout occurred.
147    KeepAlive,
148    /// Write backpressure is currently active.
149    WriteBackpressure,
150    /// Failed to decode an incoming frame.
151    Decoder(U::Error),
152    /// The peer has disconnected.
153    PeerGone(Option<IoError>),
154}
155
156impl<U> fmt::Debug for RecvError<U>
157where
158    U: Decoder,
159    <U as Decoder>::Error: fmt::Debug,
160{
161    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
162        match *self {
163            RecvError::KeepAlive => {
164                write!(fmt, "RecvError::KeepAlive")
165            }
166            RecvError::WriteBackpressure => {
167                write!(fmt, "RecvError::WriteBackpressure")
168            }
169            RecvError::Decoder(ref e) => {
170                write!(fmt, "RecvError::Decoder({e:?})")
171            }
172            RecvError::PeerGone(ref e) => {
173                write!(fmt, "RecvError::PeerGone({e:?})")
174            }
175        }
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use ntex_codec::BytesCodec;
183    use std::io;
184
185    #[test]
186    fn test_fmt() {
187        assert!(format!("{:?}", IoStatusUpdate::KeepAlive).contains("KeepAlive"));
188        assert!(format!("{:?}", RecvError::<BytesCodec>::KeepAlive).contains("KeepAlive"));
189        assert!(
190            format!("{:?}", RecvError::<BytesCodec>::WriteBackpressure)
191                .contains("WriteBackpressure")
192        );
193        assert!(
194            format!(
195                "{:?}",
196                RecvError::<BytesCodec>::Decoder(io::Error::other("err"))
197            )
198            .contains("RecvError::Decoder")
199        );
200        assert!(
201            format!(
202                "{:?}",
203                RecvError::<BytesCodec>::PeerGone(Some(io::Error::other("err")))
204            )
205            .contains("RecvError::PeerGone")
206        );
207    }
208}
ntex_io/lib.rs

ntex_io/
lib.rs
