io: pass through IO traits for StreamReader and SinkWriter (#5941)

This commit is contained in:
nicflower
2023-08-23 17:46:39 +00:00
committed by GitHub
parent 3b79be624d
commit 59c9364689
3 changed files with 47 additions and 1 deletions
+8
View File
@@ -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)
}
}
+19 -1
View File
@@ -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)
}
}
+20
View File
@@ -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)
}
}