dnet_utils/
merge.rs

1//! Merge [Sink] and [Stream] into a `dnet` transport.
2
3use std::{
4    marker::PhantomData,
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use futures::{
10    stream::{Fuse, FusedStream},
11    Sink, Stream, StreamExt,
12};
13use pin_project::pin_project;
14
15/// Transport created from merging provided sender and receiver.
16#[pin_project]
17pub struct MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
18where
19    Receiver: Stream<Item = Result<Incoming, Error>>,
20    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
21    Error: std::error::Error,
22{
23    #[pin]
24    receiver: Fuse<Receiver>,
25    #[pin]
26    sender: Sender,
27
28    #[cfg(feature = "logging")]
29    logger: dnet_base::Logger,
30
31    _incoming: PhantomData<Incoming>,
32    _outgoing: PhantomData<Outgoing>,
33}
34
35impl<Receiver, Sender, Incoming, Outgoing, Error>
36    MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
37where
38    Receiver: Stream<Item = Result<Incoming, Error>>,
39    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
40    Error: std::error::Error,
41{
42    /// Create new transport wrapping provided sender and receiver.
43    pub fn new(sender: Sender, receiver: Receiver) -> Self {
44        MergedTransport {
45            receiver: receiver.fuse(),
46            sender,
47
48            #[cfg(feature = "logging")]
49            logger: dnet_base::Logger::new::<Self>(),
50
51            _incoming: PhantomData,
52            _outgoing: PhantomData,
53        }
54    }
55}
56
57impl<Receiver, Sender, Incoming, Outgoing, Error> Sink<Outgoing>
58    for MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
59where
60    Receiver: Stream<Item = Result<Incoming, Error>>,
61    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
62    Error: std::error::Error,
63{
64    type Error = dnet_base::Error<Error>;
65
66    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
67        let me = self.project();
68        let result = me.sender.poll_ready(cx);
69
70        #[cfg(feature = "logging")]
71        me.logger.log_ready(&result);
72
73        result
74    }
75
76    fn start_send(self: Pin<&mut Self>, item: Outgoing) -> Result<(), Self::Error> {
77        let me = self.project();
78        let result = me.sender.start_send(item);
79
80        #[cfg(feature = "logging")]
81        match &result {
82            Ok(_) => me.logger.log_message_preparation_success::<Outgoing>(None),
83            Err(error) => me.logger.log_sending_failure(error),
84        }
85
86        result
87    }
88
89    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
90        let me = self.project();
91        let result = me.sender.poll_flush(cx);
92
93        #[cfg(feature = "logging")]
94        me.logger.log_flush(&result);
95
96        result
97    }
98
99    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
100        let me = self.project();
101        let result = me.sender.poll_close(cx);
102
103        #[cfg(feature = "logging")]
104        me.logger.log_close(&result);
105
106        result
107    }
108}
109
110impl<Receiver, Sender, Incoming, Outgoing, Error> Stream
111    for MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
112where
113    Receiver: Stream<Item = Result<Incoming, Error>>,
114    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
115    Error: std::error::Error,
116{
117    type Item = Result<Incoming, Error>;
118
119    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
120        let me = self.project();
121        let result = me.receiver.poll_next(cx);
122
123        #[cfg(feature = "logging")]
124        me.logger.log_receiving(&result, None);
125
126        result
127    }
128}
129
130impl<Receiver, Sender, Incoming, Outgoing, Error> FusedStream
131    for MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
132where
133    Receiver: Stream<Item = Result<Incoming, Error>>,
134    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
135    Error: std::error::Error,
136{
137    fn is_terminated(&self) -> bool {
138        self.receiver.is_terminated()
139    }
140}
141
142#[cfg(feature = "logging")]
143impl<Receiver, Sender, Incoming, Outgoing, Error> dnet_base::Logging
144    for MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
145where
146    Receiver: Stream<Item = Result<Incoming, Error>>,
147    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
148    Error: std::error::Error,
149{
150    const KIND: &'static str = "Merged";
151
152    fn with_logger<F, R>(&self, f: F) -> R
153    where
154        F: FnOnce(&dnet_base::Logger) -> R,
155    {
156        f(&self.logger)
157    }
158
159    fn with_logger_mut<F, R>(&mut self, f: F) -> R
160    where
161        F: FnOnce(&mut dnet_base::Logger) -> R,
162    {
163        f(&mut self.logger)
164    }
165}
166
167/// Merge provided sender and receiver into a single transport.
168pub fn merge<Receiver, Sender, Incoming, Outgoing, Error>(
169    sender: Sender,
170    receiver: Receiver,
171) -> MergedTransport<Receiver, Sender, Incoming, Outgoing, Error>
172where
173    Receiver: Stream<Item = Result<Incoming, Error>>,
174    Sender: Sink<Outgoing, Error = dnet_base::Error<Error>>,
175    Error: std::error::Error,
176{
177    MergedTransport::new(sender, receiver)
178}
dnet_utils/merge.rs

dnet_utils/
merge.rs
