veilid_tools/
clone_stream.rs

1use super::*;
2
3use core::pin::Pin;
4use core::task::{Context, Poll};
5use futures_util::AsyncRead as Read;
6use futures_util::AsyncWrite as Write;
7use futures_util::Sink;
8use futures_util::Stream;
9use std::io;
10
11pub struct CloneStream<T>
12where
13    T: Unpin,
14{
15    inner: Arc<Mutex<T>>,
16}
17
18impl<T> Clone for CloneStream<T>
19where
20    T: Unpin,
21{
22    fn clone(&self) -> Self {
23        Self {
24            inner: self.inner.clone(),
25        }
26    }
27}
28
29impl<T> CloneStream<T>
30where
31    T: Unpin,
32{
33    pub fn new(t: T) -> Self {
34        Self {
35            inner: Arc::new(Mutex::new(t)),
36        }
37    }
38}
39
40impl<T> Read for CloneStream<T>
41where
42    T: Read + Unpin,
43{
44    fn poll_read(
45        self: Pin<&mut Self>,
46        cx: &mut Context<'_>,
47        buf: &mut [u8],
48    ) -> Poll<io::Result<usize>> {
49        let mut inner = self.inner.lock();
50        Pin::new(&mut *inner).poll_read(cx, buf)
51    }
52}
53
54impl<T> Write for CloneStream<T>
55where
56    T: Write + Unpin,
57{
58    fn poll_write(
59        self: Pin<&mut Self>,
60        cx: &mut Context<'_>,
61        buf: &[u8],
62    ) -> Poll<io::Result<usize>> {
63        let mut inner = self.inner.lock();
64        Pin::new(&mut *inner).poll_write(cx, buf)
65    }
66
67    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
68        let mut inner = self.inner.lock();
69        Pin::new(&mut *inner).poll_flush(cx)
70    }
71
72    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
73        let mut inner = self.inner.lock();
74        Pin::new(&mut *inner).poll_close(cx)
75    }
76}
77
78impl<T> Stream for CloneStream<T>
79where
80    T: Stream + Unpin,
81{
82    type Item = T::Item;
83
84    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
85        let mut inner = self.inner.lock();
86        Pin::new(&mut *inner).poll_next(cx)
87    }
88}
89
90impl<T, Item> Sink<Item> for CloneStream<T>
91where
92    T: Sink<Item> + Unpin,
93{
94    type Error = T::Error;
95
96    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
97        let mut inner = self.inner.lock();
98        Pin::new(&mut *inner).poll_ready(cx)
99    }
100    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error> {
101        let mut inner = self.inner.lock();
102        Pin::new(&mut *inner).start_send(item)
103    }
104    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
105        let mut inner = self.inner.lock();
106        Pin::new(&mut *inner).poll_flush(cx)
107    }
108    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
109        let mut inner = self.inner.lock();
110        Pin::new(&mut *inner).poll_close(cx)
111    }
112}
veilid_tools/clone_stream.rs

veilid_tools/
clone_stream.rs
