//! Extension trait for `Sink`. use std::{ marker::PhantomData, pin::Pin, task::{Context, Poll}, }; use futures::{ready, sink::Sink}; use pin_project::pin_project; /// Extension trait for `Sink` pub trait SinkExt: Sink { /// As `Sink::with`, but takes a function that returns an `Item` rather /// than `Future`. fn with_fn(self, func: F) -> WithFn // or error? where Self: Sized, F: FnMut(T) -> Result, E: From, { WithFn { sink: self, func, _phantom: PhantomData, } } } impl SinkExt for S where S: Sink {} /// Sink returned by [`SinkExt::with_fn`]. #[pin_project] pub struct WithFn { /// The underlying sink #[pin] sink: S, /// The user-provided function. func: F, /// Phantom data to ensure type consistency. _phantom: PhantomData Result>, } impl Sink for WithFn where S: Sink, F: FnMut(T) -> Result, E: From, { type Error = E; fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { ready!(self.project().sink.poll_ready(cx))?; Poll::Ready(Ok(())) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { ready!(self.project().sink.poll_flush(cx))?; Poll::Ready(Ok(())) } fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { ready!(self.project().sink.poll_close(cx))?; Poll::Ready(Ok(())) } fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> { let this = self.project(); let item = (this.func)(item)?; this.sink.start_send(item).map_err(E::from) } }