diff --git a/tokio-util/src/io/copy_to_bytes.rs b/tokio-util/src/io/copy_to_bytes.rs
index 9509e7119..f0b5c3579 100644
--- a/tokio-util/src/io/copy_to_bytes.rs
+++ b/tokio-util/src/io/copy_to_bytes.rs
@@ -1,4 +1,5 @@
 use bytes::Bytes;
+use futures_core::stream::Stream;
 use futures_sink::Sink;
 use pin_project_lite::pin_project;
 use std::pin::Pin;
@@ -66,3 +67,10 @@ where
         self.project().inner.poll_close(cx)
     }
 }
+
+impl<S: Stream> Stream for CopyToBytes<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)
+    }
+}
diff --git a/tokio-util/src/io/sink_writer.rs b/tokio-util/src/io/sink_writer.rs
index f62935dd2..e07895287 100644
--- a/tokio-util/src/io/sink_writer.rs
+++ b/tokio-util/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`].
@@ -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)
+    }
+}
diff --git a/tokio-util/src/io/stream_reader.rs b/tokio-util/src/io/stream_reader.rs
index 3353722c5..d30f7f0f4 100644
--- a/tokio-util/src/io/stream_reader.rs
+++ b/tokio-util/src/io/stream_reader.rs
@@ -1,5 +1,6 @@
 use bytes::Buf;
 use futures_core::stream::Stream;
+use futures_sink::Sink;
 use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -324,3 +325,22 @@ impl<S, B> StreamReader<S, B> {
         }
     }
 }
+
+impl<S: Sink<T, Error = E>, E, T> Sink<T> for StreamReader<S, E> {
+    type Error = E;
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_ready(cx)
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+        self.project().inner.start_send(item)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_flush(cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_close(cx)
+    }
+}