use crate::prelude::*;
#[derive(Debug)]
pub struct MapError<C, Fun> {
inner: C,
fun: Option<Fun>,
}
impl<C, Fun> MapError<C, Fun> {
pub(crate) fn new(inner: C, fun: Fun) -> Self {
Self {
inner,
fun: Some(fun),
}
}
pub fn into_inner(self) -> C {
self.inner
}
}
impl<C, Fun> AsRef<C> for MapError<C, Fun> {
fn as_ref(&self) -> &C {
&self.inner
}
}
impl<C, Fun> AsMut<C> for MapError<C, Fun> {
fn as_mut(&mut self) -> &mut C {
&mut self.inner
}
}
impl<C, Fun, NewErr> Consumer for MapError<C, Fun>
where
C: Consumer,
Fun: FnOnce(C::Error) -> NewErr,
{
type Item = C::Item;
type Final = C::Final;
type Error = NewErr;
async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
self.inner.consume(val).await.map_err(|err| {
(self
.fun
.take()
.expect("Must not use a consumer after it emitted an error"))(err)
})
}
async fn flush(&mut self) -> Result<(), Self::Error> {
self.inner.flush().await.map_err(|err| {
(self
.fun
.take()
.expect("Must not use a consumer after it emitted an error"))(err)
})
}
}
impl<C, Fun, NewErr> BulkConsumer for MapError<C, Fun>
where
C: BulkConsumer,
Fun: FnOnce(C::Error) -> NewErr,
{
async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
where
F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
{
self.inner.expose_slots_gracefully(f).await.map_err(|err| {
(
err.0,
self.fun
.take()
.expect("Must not use a consumer after it emitted an error")(
err.1
),
)
})
}
}
#[derive(Debug)]
pub struct MapAsyncError<C, Fun> {
inner: C,
fun: Option<Fun>,
}
impl<C, Fun> MapAsyncError<C, Fun> {
pub(crate) fn new(inner: C, fun: Fun) -> Self {
Self {
inner,
fun: Some(fun),
}
}
pub fn into_inner(self) -> C {
self.inner
}
}
impl<C, Fun> AsRef<C> for MapAsyncError<C, Fun> {
fn as_ref(&self) -> &C {
&self.inner
}
}
impl<C, Fun> AsMut<C> for MapAsyncError<C, Fun> {
fn as_mut(&mut self) -> &mut C {
&mut self.inner
}
}
impl<C, Fun, NewErr> Consumer for MapAsyncError<C, Fun>
where
C: Consumer,
Fun: AsyncFnOnce(C::Error) -> NewErr,
{
type Item = C::Item;
type Final = C::Final;
type Error = NewErr;
async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
match self.inner.consume(val).await {
Ok(yay) => Ok(yay),
Err(err) => Err((self
.fun
.take()
.expect("Must not use a consumer after it emitted an error"))(
err
)
.await),
}
}
async fn flush(&mut self) -> Result<(), Self::Error> {
match self.inner.flush().await {
Ok(yay) => Ok(yay),
Err(err) => Err((self
.fun
.take()
.expect("Must not use a consumer after it emitted an error"))(
err
)
.await),
}
}
}
impl<C, Fun, NewErr> BulkConsumer for MapAsyncError<C, Fun>
where
C: BulkConsumer,
Fun: AsyncFnOnce(C::Error) -> NewErr,
{
async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
where
F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
{
match self.inner.expose_slots_gracefully(f).await {
Ok(yay) => Ok(yay),
Err(err) => Err((
err.0,
self.fun
.take()
.expect("Must not use a consumer after it emitted an error")(
err.1
)
.await,
)),
}
}
}