ufotofu 0.10.1

use crate::prelude::*;

/// A wrapper for a pair of [`Consumer`]s, which forwards consumed values to the first consumer, until the first consumer reports an error. From that point on, values are forwarded to the second consumer.
///
/// When the wrapper is closed by passing the [`final`](Consumer::Final) value of the second consumer, the first consumer is flushed if it has not reported an error.
///
/// Created via [`ConsumerExt::to_chain`].
///
/// <br/>Counterpart: the [`producer::Chain`] type.
#[derive(Debug, PartialEq)]
pub struct Chain<P, Q> {
    chained: (P, Q),
    first_exhausted: bool,
}

impl<P, Q> Chain<P, Q> {
    pub(crate) fn new(first: P, second: Q) -> Self {
        Self {
            chained: (first, second),
            first_exhausted: false,
        }
    }

    /// Consumes `self` to return its two wrapped [`consumer`](Consumer)s.
    pub fn into_inner(self) -> (P, Q) {
        self.chained
    }

    /// Returns `true` if the first [`Consumer`] in the chain has reported an error, otherwise `false`.
    pub fn first_exhausted(&self) -> bool {
        self.first_exhausted
    }

    async fn try_flush_first_consumer(&mut self)
    where
        P: Consumer,
    {
        if !self.first_exhausted && self.chained.0.flush().await.is_err() {
            self.first_exhausted = true;
        }
    }
}

impl<P, Q> AsRef<(P, Q)> for Chain<P, Q> {
    fn as_ref(&self) -> &(P, Q) {
        &self.chained
    }
}

impl<P, Q> Consumer for Chain<P, Q>
where
    P: Consumer<Error = ()>,
    Q: Consumer<Item = P::Item>,
    P::Item: Copy,
{
    type Item = P::Item;

    type Final = Q::Final;

    type Error = Q::Error;

    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        match val {
            Right(finisher) => {
                self.try_flush_first_consumer().await;
                return self.chained.1.consume(Right(finisher)).await;
            }
            Left(item) => {
                if !self.first_exhausted {
                    if self.chained.0.consume(Left(item)).await.is_err() {
                        self.first_exhausted = true;
                    } else {
                        return Ok(());
                    }
                }
                self.chained.1.consume(Left(item)).await
            }
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        self.try_flush_first_consumer().await;
        if self.first_exhausted {
            self.chained.1.flush().await?;
        }
        Ok(())
    }
}

impl<P, Q> BulkConsumer for Chain<P, Q>
where
    P: BulkConsumer<Error = ()>,
    Q: BulkConsumer<Item = P::Item>,
    P::Item: Copy,
{
    async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
    where
        F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
    {
        if self.first_exhausted {
            return self.chained.1.expose_slots_gracefully(f).await;
        }

        match self.chained.0.expose_slots_gracefully(f).await {
            Ok(yay) => Ok(yay),
            Err((f, ())) => {
                self.first_exhausted = true;
                self.chained.1.expose_slots_gracefully(f).await
            }
        }
    }
}

/// A wrapper for a pair of [`Consumer`]s, which clones and forwards consumed values to the first consumer, until the first consumer reports an error. From that point on, values are forwarded to the second consumer.
///
/// When the wrapper is closed by passing the [`final`](Consumer::Final) value of the second consumer, the first consumer is flushed if it has not reported an error.
///
/// Created via [`ConsumerExt::to_cloning_chain`]. For types which implement [`Copy`], prefer [`ConsumerExt::to_chain`] to avoid the overhead of cloning items.
///
/// <br/>Counterpart: the [`producer::Chain`] type.
#[derive(Debug, PartialEq)]
pub struct CloningChain<P, Q> {
    chained: (P, Q),
    first_exhausted: bool,
}

impl<P, Q> CloningChain<P, Q> {
    pub(crate) fn new(first: P, second: Q) -> Self {
        Self {
            chained: (first, second),
            first_exhausted: false,
        }
    }

    /// Consumes `self` to return its two wrapped [`consumer`](Consumer)s.
    pub fn into_inner(self) -> (P, Q) {
        self.chained
    }

    /// Returns `true` if the first [`Consumer`] in the chain has reported an error, otherwise `false`.
    pub fn first_exhausted(&self) -> bool {
        self.first_exhausted
    }

    async fn try_flush_first_consumer(&mut self)
    where
        P: Consumer,
    {
        if !self.first_exhausted && self.chained.0.flush().await.is_err() {
            self.first_exhausted = true;
        }
    }
}

impl<P, Q> AsRef<(P, Q)> for CloningChain<P, Q> {
    fn as_ref(&self) -> &(P, Q) {
        &self.chained
    }
}

impl<P, Q> Consumer for CloningChain<P, Q>
where
    P: Consumer<Error = ()>,
    Q: Consumer<Item = P::Item>,
    P::Item: Clone,
{
    type Item = P::Item;

    type Final = Q::Final;

    type Error = Q::Error;

    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        match val {
            Right(finisher) => {
                self.try_flush_first_consumer().await;
                return self.chained.1.consume(Right(finisher)).await;
            }
            Left(item) => {
                if !self.first_exhausted {
                    if self.chained.0.consume(Left(item.clone())).await.is_err() {
                        self.first_exhausted = true;
                    } else {
                        return Ok(());
                    }
                }
                self.chained.1.consume(Left(item)).await
            }
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        self.try_flush_first_consumer().await;
        if self.first_exhausted {
            self.chained.1.flush().await?;
        }
        Ok(())
    }
}

impl<P, Q> BulkConsumer for CloningChain<P, Q>
where
    P: BulkConsumer<Error = ()>,
    Q: BulkConsumer<Item = P::Item>,
    P::Item: Clone,
{
    async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
    where
        F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
    {
        if self.first_exhausted {
            return self.chained.1.expose_slots_gracefully(f).await;
        }

        match self.chained.0.expose_slots_gracefully(f).await {
            Ok(yay) => Ok(yay),
            Err((f, ())) => {
                self.first_exhausted = true;
                self.chained.1.expose_slots_gracefully(f).await
            }
        }
    }
}