aboutsummaryrefslogtreecommitdiff
path: root/src/io/sink_writer.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/io/sink_writer.rs')
-rw-r--r--src/io/sink_writer.rs22
1 files changed, 20 insertions, 2 deletions
diff --git a/src/io/sink_writer.rs b/src/io/sink_writer.rs
index f2af262..e078952 100644
--- a/src/io/sink_writer.rs
+++ b/src/io/sink_writer.rs
@@ -1,11 +1,12 @@
use futures_core::ready;
use futures_sink::Sink;
+use futures_core::stream::Stream;
use pin_project_lite::pin_project;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
-use tokio::io::AsyncWrite;
+use tokio::io::{AsyncRead, AsyncWrite};
pin_project! {
/// Convert a [`Sink`] of byte chunks into an [`AsyncWrite`].
@@ -59,7 +60,7 @@ pin_project! {
/// [`CopyToBytes`]: crate::io::CopyToBytes
/// [`Encoder`]: crate::codec::Encoder
/// [`Sink`]: futures_sink::Sink
- /// [`codec`]: tokio_util::codec
+ /// [`codec`]: crate::codec
#[derive(Debug)]
pub struct SinkWriter<S> {
#[pin]
@@ -115,3 +116,20 @@ where
self.project().inner.poll_close(cx).map_err(Into::into)
}
}
+
+impl<S: Stream> Stream for SinkWriter<S> {
+ type Item = S::Item;
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ self.project().inner.poll_next(cx)
+ }
+}
+
+impl<S: AsyncRead> AsyncRead for SinkWriter<S> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut tokio::io::ReadBuf<'_>,
+ ) -> Poll<io::Result<()>> {
+ self.project().inner.poll_read(cx, buf)
+ }
+}