Hi all!
I have a slightly advanced nugget for you. I'm currently working on a PR for tokio and think I'm pretty much done, but the test shows that I am still missing something essential.
As a background, I am trying to implement Asyncwrite for generic Sinks which can cast their types to &[u8] and have sink error type std::io::Error (like a file-like interface). My plan was to build up these conversions in the struct and then implement AsyncWriteon my struct, assuming it has such a sink type wrapped.
I seem to make a logic error somewhere in there. It's somewhat intricate though - because I get rust-analyzer hints that AsyncWriteExt is indeed implemented by struct, but if I want to run cargo test, I get complaints that my type doesn't implement it. I have already been thinking that AsyncWriteExt requires the type to be ?Sized and I couldn't tell whether that was the case or not. I'm running a bit out of ideas and would appreciate what I'm missing here.
The following code block contains both the implementation and the test case that fails.
///
/// PART 1: DEFINITION AND TRAIT IMPL
///
use futures_sink::Sink;
use futures_util::{
future::{ok, Ready},
sink::{SinkMapErr, With},
SinkExt,
};
use pin_project_lite::pin_project;
use std::io;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncWrite;
pin_project! {
/// Convert a [`Sink`] of byte chunks into an [`AsyncWrite`].
/// Bytes are sent into the sink and the sink is flushed once all bytes are sent. If an error occurs during the sending progress,
/// the number of sent but unflushed bytes are saved in case the flushing operation stays unsuccessful.
/// For the inverse operation of defining an [`AsyncWrite`] from a [`Sink`] you need to define a [`codec`].
///
/// # Example
///
/// ```
/// use futures_util::SinkExt;
/// use std::io::{Error, ErrorKind};
/// use tokio::io::AsyncWriteExt;
/// use tokio_util::io::SinkWriter;
/// use tokio_util::sync::PollSender;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Error> {
/// // Construct a channel pair to send data across and wrap a pollable sink.
/// // Note that the sink must mimic a writable object, e.g. have `std::io::Error`
/// // as its error type.
/// let (tx, mut rx) = tokio::sync::mpsc::channel::<&[u8]>(10);
/// let mut writer = SinkWriter::new(
/// PollSender::new(tx).sink_map_err(|_| Error::from(ErrorKind::Other)),
/// );
/// // Write data to our interface...
/// let data: [u8; 4] = [1, 2, 3, 4];
/// let _ = writer.write(&data).await?;
/// // ... and receive it.
/// assert_eq!(&data, rx.recv().await.unwrap());
///
/// # Ok(())
/// # }
/// ```
///
///
/// [`AsyncWrite`]: tokio::io::AsyncWrite
/// [`Sink`]: futures_sink::Sink
/// [`codec`]: tokio_util::codec
#[derive(Debug)]
pub struct SinkWriter<S, T, E>
{
#[pin]
inner: S,
_type_marker: PhantomData<T>,
_error_marker: PhantomData<E>,
}
}
impl<S, E, T>
SinkWriter<
SinkMapErr<
With<S, T, &[u8], Ready<Result<T, E>>, for<'r> fn(&'r [u8]) -> Ready<Result<T, E>>>,
fn(E) -> io::Error,
>,
T,
E,
>
where
S: Sink<T>,
E: Into<io::Error> + From<<S as Sink<T>>::Error>,
T: for<'b> From<&'b [u8]> + ?Sized,
{
/// Creates a new [`SinkWriter`].
pub fn new(sink: S) -> Self {
Self {
inner: sink
.with(Self::sink_type_coupler as _)
.sink_map_err(Self::sink_error_coupler as _),
_type_marker: PhantomData,
_error_marker: PhantomData,
}
}
/// Anonymous helper function which provides an awaitable function that
/// wraps the provided sink into a [`Sink<&[u8]>`].
fn sink_type_coupler(buf: &[u8]) -> Ready<Result<T, E>> {
ok::<T, E>(From::from(buf))
}
/// Anonymous helper function which provides an awaitable function that
/// wraps the provided sink erro into a [`io::Error`].
fn sink_error_coupler(e: E) -> io::Error {
e.into()
}
/// Gets a reference to the underlying sink.
///
/// It is inadvisable to directly write to the underlying sink.
pub fn get_ref(&self) -> &S {
self.inner.get_ref().get_ref()
}
/// Gets a mutable reference to the underlying sink.
///
/// It is inadvisable to directly write to the underlying sink.
pub fn get_mut(&mut self) -> &mut S {
self.inner.get_mut().get_mut()
}
/// Consumes this `SinkWriter`, returning the underlying sink.
///
/// It is inadvisable to directly write to the underlying sink.
pub fn into_inner(self) -> S {
self.inner.into_inner().into_inner()
}
}
impl<S, E, T> AsyncWrite for SinkWriter<S, T, E>
where
for<'r> S: Sink<&'r [u8], Error = io::Error>,
{
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, io::Error>> {
match self.as_mut().project().inner.poll_ready(cx) {
Poll::Ready(Ok(())) => {
if let Err(e) = self.as_mut().project().inner.start_send(buf) {
Poll::Ready(Err(e))
} else {
Poll::Ready(Ok(buf.len()))
}
}
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => {
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
self.project().inner.poll_flush(cx).map_err(Into::into)
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
self.project().inner.poll_close(cx).map_err(Into::into)
}
}
///
/// PART 2, THE TEST
///
#![warn(rust_2018_idioms)]
use futures_util::SinkExt;
use std::fmt::{self, Debug, Display};
use std::io::{self, Error, ErrorKind};
use tokio::io::AsyncWriteExt;
use tokio_util::io::SinkWriter;
use tokio_util::sync::{PollSendError, PollSender};
#[derive(Debug)]
struct PollSendErrorCoupler<T>(PollSendError<T>);
impl<T: Debug> Display for PollSendErrorCoupler<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
Display::fmt(&self, f)
}
}
impl<T: Debug> std::error::Error for PollSendErrorCoupler<T> {
}
impl<T> From<PollSendError<T>> for PollSendErrorCoupler<T> {
fn from(e: PollSendError<T>) -> Self {
PollSendErrorCoupler(e)
}
}
impl<T> Into<io::Error> for PollSendErrorCoupler<T> {
fn into(self) -> io::Error {
io::Error::from(ErrorKind::BrokenPipe)
}
}
#[tokio::test]
async fn test_sink_writer() -> Result<(), Error> {
// Construct a channel pair to send data across and wrap a pollable sink.
// Note that the sink must mimic a writable object, e.g. have `std::io::Error`
// as its error type.
let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(4);
let writer: SinkWriter<_, _, PollSendErrorCoupler<Vec<u8>>> =
SinkWriter::new(PollSender::new(tx));
// Write data to our interface...
let data: [u8; 4] = [1, 2, 3, 4];
let _ = writer.write(&data).await; // <- This should work but doesn't....
// ... and receive it.
assert_eq!(&data, rx.recv().await.unwrap().as_slice());
Ok(())
}
I have a suspicion why the trait is not implemented, I don't think the implementation should be fulfilled by any S and T, so I was also trying around with headers like this one, which fails since S is unrestricted:
impl<W, S, E, T> AsyncWrite for SinkWriter<S, T, E>
where
for<'r> S: Sink<&'r [u8], Error = io::Error>,
W: Sink<T>,
E: Into<io::Error> + From<<W as Sink<T>>::Error>,
T: for<'b> From<&'b [u8]> + ?Sized,
{
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
[...]
But I'm not sure.
Any help would be appreciated a lot!