dnet_utils/
wall.rs

1//! Transport that rejects every operation.
2
3use std::{
4    fmt::Display,
5    marker::PhantomData,
6    pin::Pin,
7    task::{Context, Poll},
8};
9
10use futures::{stream::FusedStream, Sink, Stream};
11
12/// Transport rejecting every operation.
13///
14/// Attempting to send, receive from, flush or close this transport will result in an error.
15pub struct Wall<Incoming, Outgoing> {
16    #[cfg(feature = "logging")]
17    logger: dnet_base::Logger,
18
19    _incoming: PhantomData<Incoming>,
20    _outgoing: PhantomData<Outgoing>,
21}
22
23impl<Incoming, Outgoing> Default for Wall<Incoming, Outgoing> {
24    fn default() -> Self {
25        Self {
26            #[cfg(feature = "logging")]
27            logger: dnet_base::Logger::new::<Self>(),
28
29            _incoming: PhantomData,
30            _outgoing: PhantomData,
31        }
32    }
33}
34
35/// Transport-specific errors for [`Wall`].
36#[derive(Debug, PartialEq, Eq)]
37pub enum Error {
38    /// Attempted to send message into the wall.
39    Send,
40
41    /// Attempted to receive message from the wall.
42    Receive,
43
44    /// Attempted to flush the wall.
45    Flush,
46
47    /// Attempted to close the wall.
48    Close,
49}
50
51impl Display for Error {
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        match self {
54            Self::Send => write!(f, "attempted to send message into the wall"),
55            Self::Receive => write!(f, "attempted to receive message from the wall"),
56            Self::Flush => write!(f, "attempted to flush the wall"),
57            Self::Close => write!(f, "attempted to close the wall"),
58        }
59    }
60}
61
62impl std::error::Error for Error {}
63
64impl<Incoming, Outgoing> Sink<Outgoing> for Wall<Incoming, Outgoing> {
65    type Error = dnet_base::Error<Error>;
66
67    fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
68        let result = Poll::Ready(Ok(()));
69
70        #[cfg(feature = "logging")]
71        self.logger.log_ready(&result);
72
73        result
74    }
75
76    fn start_send(self: Pin<&mut Self>, _item: Outgoing) -> Result<(), Self::Error> {
77        let result = Err(dnet_base::Error::Other(Error::Send));
78
79        #[cfg(feature = "logging")]
80        match &result {
81            Ok(_) => unreachable!(),
82            Err(error) => self.logger.log_sending_failure(error),
83        }
84
85        result
86    }
87
88    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
89        let result = Poll::Ready(Err(dnet_base::Error::Other(Error::Flush)));
90
91        #[cfg(feature = "logging")]
92        self.logger.log_flush(&result);
93
94        result
95    }
96
97    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
98        let result = Poll::Ready(Err(dnet_base::Error::Other(Error::Close)));
99
100        #[cfg(feature = "logging")]
101        self.logger.log_close(&result);
102
103        result
104    }
105}
106
107impl<Incoming, Outgoing> Stream for Wall<Incoming, Outgoing> {
108    type Item = Result<Incoming, Error>;
109
110    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
111        let result = Poll::Ready(Some(Err(Error::Receive)));
112
113        #[cfg(feature = "logging")]
114        self.logger.log_receiving(&result, None);
115
116        result
117    }
118}
119
120impl<Incoming, Outgoing> FusedStream for Wall<Incoming, Outgoing> {
121    fn is_terminated(&self) -> bool {
122        false
123    }
124}
125
126#[cfg(feature = "logging")]
127impl<Incoming, Outgoing> dnet_base::Logging for Wall<Incoming, Outgoing> {
128    const KIND: &'static str = "Wall";
129
130    fn with_logger<F, R>(&self, f: F) -> R
131    where
132        F: FnOnce(&dnet_base::Logger) -> R,
133    {
134        f(&self.logger)
135    }
136
137    fn with_logger_mut<'a, F, R>(&mut self, f: F) -> R
138    where
139        F: FnOnce(&mut dnet_base::Logger) -> R,
140    {
141        f(&mut self.logger)
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use dnet_base::Receive;
148    use dnet_tests::{dtest, dtest_configure};
149    use futures::SinkExt;
150
151    use crate::wall::{Error, Wall};
152
153    dtest_configure!();
154
155    fn enable_logging<Incoming, Outgoing>(wall: &mut Wall<Incoming, Outgoing>) {
156        #[cfg(not(feature = "logging"))]
157        {
158            let _ = wall;
159        }
160
161        #[cfg(feature = "logging")]
162        {
163            use dnet_base::Logging;
164            dnet_tests::init_subscriber();
165            wall.enable_logging();
166        }
167    }
168
169    #[dtest]
170    async fn test_wall_send_error() {
171        let mut wall: Wall<i32, &str> = Wall::default();
172
173        enable_logging(&mut wall);
174
175        let result = wall.send("Hello").await;
176        assert_eq!(result, Err(dnet_base::Error::Other(Error::Send)));
177    }
178
179    #[dtest]
180    async fn test_wall_flush_error() {
181        let mut wall: Wall<i32, &str> = Wall::default();
182
183        enable_logging(&mut wall);
184
185        let result = wall.flush().await;
186        assert_eq!(result, Err(dnet_base::Error::Other(Error::Flush)));
187    }
188
189    #[dtest]
190    async fn test_wall_close_error() {
191        let mut wall: Wall<i32, &str> = Wall::default();
192
193        enable_logging(&mut wall);
194
195        let result = wall.close().await;
196        assert_eq!(result, Err(dnet_base::Error::Other(Error::Close)));
197    }
198
199    #[dtest]
200    async fn test_wall_receive_error() {
201        let mut wall: Wall<i32, &str> = Wall::default();
202
203        enable_logging(&mut wall);
204
205        let result = wall.receive().await;
206        assert_eq!(result, Err(dnet_base::Error::Other(Error::Receive)));
207    }
208}
dnet_utils/wall.rs

dnet_utils/
wall.rs
