dnet_utils/
channel.rs

1//! Transport for communication over
2//! [futures](https://github.com/rust-lang/futures-rs) channels.
3//!
4//! Useful for testing and debugging.
5//!
6//! ## Example
7//!
8//! ```ignore
9//! let (mut left, mut right) = transports();
10//!
11//! left.send("Hello World!").await.unwrap();
12//! right.send(123).await.unwrap();
13//!
14//! use dnet::Receive;
15//! assert_eq!(right.receive().await.unwrap(), "Hello World!");
16//! assert_eq!(left.receive().await.unwrap(), 123);
17//! ```
18
19use std::{
20    fmt::Display,
21    pin::Pin,
22    task::{Context, Poll},
23};
24
25use futures::{
26    channel::mpsc::{unbounded, SendError, UnboundedReceiver, UnboundedSender},
27    sink::SinkMapErr,
28    stream::{FusedStream, Map},
29    Sink, SinkExt, Stream, StreamExt,
30};
31use pin_project::pin_project;
32
33use crate::merge::{merge, MergedTransport};
34
35/// Channel transport error.
36#[derive(Debug)]
37pub enum Error {
38    /// Channel is full.
39    ChannelIsFull,
40}
41
42impl Display for Error {
43    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44        write!(f, "channel is full")
45    }
46}
47
48impl std::error::Error for Error {}
49
50type Receiver<T> = Map<UnboundedReceiver<T>, fn(T) -> Result<T, Error>>;
51type Sender<T> = SinkMapErr<UnboundedSender<T>, fn(SendError) -> dnet_base::Error<Error>>;
52
53/// Channel transport.
54#[pin_project]
55pub struct ChannelTransport<Incoming, Outgoing> {
56    #[pin]
57    inner: MergedTransport<Receiver<Incoming>, Sender<Outgoing>, Incoming, Outgoing, Error>,
58}
59
60impl<Incoming, Outgoing> Sink<Outgoing> for ChannelTransport<Incoming, Outgoing> {
61    type Error = dnet_base::Error<Error>;
62
63    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
64        let me = self.project();
65        me.inner.poll_ready(cx)
66    }
67
68    fn start_send(self: Pin<&mut Self>, item: Outgoing) -> Result<(), Self::Error> {
69        let me = self.project();
70        me.inner.start_send(item)
71    }
72
73    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
74        let me = self.project();
75        me.inner.poll_flush(cx)
76    }
77
78    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
79        let me = self.project();
80        me.inner.poll_close(cx)
81    }
82}
83
84impl<Incoming, Outgoing> Stream for ChannelTransport<Incoming, Outgoing> {
85    type Item = Result<Incoming, Error>;
86
87    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
88        let me = self.project();
89        me.inner.poll_next(cx)
90    }
91}
92
93impl<Incoming, Outgoing> FusedStream for ChannelTransport<Incoming, Outgoing> {
94    fn is_terminated(&self) -> bool {
95        self.inner.is_terminated()
96    }
97}
98
99impl<Incoming, Outgoing> From<(UnboundedSender<Outgoing>, UnboundedReceiver<Incoming>)>
100    for ChannelTransport<Incoming, Outgoing>
101{
102    fn from(pair: (UnboundedSender<Outgoing>, UnboundedReceiver<Incoming>)) -> Self {
103        let sender = pair
104            .0
105            .sink_map_err(map_error as fn(SendError) -> dnet_base::Error<Error>);
106        let receiver = pair.1.map(map as fn(Incoming) -> Result<Incoming, Error>);
107
108        #[allow(unused_mut)]
109        let mut inner = merge(sender, receiver);
110        #[cfg(feature = "logging")]
111        {
112            use dnet_base::logging::Logging;
113            inner.with_logger_mut(|logger| logger.override_kind::<Self>())
114        }
115
116        ChannelTransport { inner }
117    }
118}
119
120#[cfg(feature = "logging")]
121impl<Incoming, Outgoing> dnet_base::Logging for ChannelTransport<Incoming, Outgoing> {
122    const KIND: &'static str = "Channel";
123
124    fn with_logger<F, R>(&self, f: F) -> R
125    where
126        F: FnOnce(&dnet_base::Logger) -> R,
127    {
128        self.inner.with_logger(f)
129    }
130
131    fn with_logger_mut<F, R>(&mut self, f: F) -> R
132    where
133        F: FnOnce(&mut dnet_base::Logger) -> R,
134    {
135        self.inner.with_logger_mut(f)
136    }
137}
138
139/// Create two connected transports over two unbounded channels.
140#[allow(clippy::type_complexity)]
141pub fn transports<A, B>() -> (ChannelTransport<A, B>, ChannelTransport<B, A>) {
142    let (left_sender, right_receiver) = unbounded();
143    let (right_sender, left_receiver) = unbounded();
144
145    let left = (left_sender, left_receiver).into();
146    let right = (right_sender, right_receiver).into();
147
148    (left, right)
149}
150
151fn map<T>(value: T) -> Result<T, Error> {
152    Ok(value)
153}
154
155fn map_error(error: SendError) -> dnet_base::Error<Error> {
156    if error.is_full() {
157        dnet_base::Error::Other(Error::ChannelIsFull)
158    } else {
159        dnet_base::Error::Closed
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use dnet_tests::{dtest, dtest_configure};
166
167    use super::transports;
168
169    dtest_configure!();
170
171    #[dtest]
172    async fn test_transport() {
173        let (left, right) = transports();
174        dnet_tests::test_transport(left, right).await;
175    }
176
177    #[dtest]
178    async fn test_unit_message() {
179        let (left, right) = transports();
180        dnet_tests::test_unit_message(left, right).await;
181    }
182
183    #[dtest]
184    async fn test_stream() {
185        let (left, right) = transports();
186        dnet_tests::test_stream(left, right).await;
187    }
188}
dnet_utils/channel.rs

dnet_utils/
channel.rs
