use super::*; use core::pin::Pin; use core::task::{Context, Poll}; use futures_util::AsyncRead as Read; use futures_util::AsyncWrite as Write; use futures_util::Sink; use futures_util::Stream; use std::io; pub struct CloneStream where T: Unpin, { inner: Arc>, } impl Clone for CloneStream where T: Unpin, { fn clone(&self) -> Self { Self { inner: self.inner.clone(), } } } impl CloneStream where T: Unpin, { pub fn new(t: T) -> Self { Self { inner: Arc::new(Mutex::new(t)), } } } impl Read for CloneStream where T: Read + Unpin, { fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_read(cx, buf) } } impl Write for CloneStream where T: Write + Unpin, { fn poll_write( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8], ) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_write(cx, buf) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_flush(cx) } fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_close(cx) } } impl Stream for CloneStream where T: Stream + Unpin, { type Item = T::Item; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_next(cx) } } impl Sink for CloneStream where T: Sink + Unpin, { type Error = T::Error; fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_ready(cx) } fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).start_send(item) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_flush(cx) } fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut inner = self.inner.lock(); Pin::new(&mut *inner).poll_close(cx) } }