ufotofu 0.12.5

//! Consumer functionality for [`Vec`].
//!
//! Specifically, the module provides
//!
//! - an [`IntoConsumer`] impl for `Vec<T>`,
//! - an [`IntoConsumer`] impl for `&mut Vec<T>`.
//!
//! <br/>Counterpart: the [`producer::compat::vec`] module.

use core::fmt::Debug;

use alloc::vec::Vec;

use crate::prelude::*;

/// The consumer of the [`IntoConsumer`] impl of `Vec<T>`; it appends consumed data to the [`Vec`].
///
/// Use the [`Into`] impl to recover the vec when you are done consuming items.
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let mut c = vec![].into_consumer();
///
/// c.consume_item(1).await?;
/// c.consume_item(2).await?;
/// c.consume_item(4).await?;
///
/// let vec: Vec<_> = c.into();
/// assert_eq!(vec, vec![1, 2, 4]);
/// # Result::<(), Infallible>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: the [producer::compat::vec::IntoProducer] type.
#[derive(Clone)]

pub struct IntoConsumer<T>(Vec<T>, usize);
// The usize is the number of items consumed so far. For bulk consumption, we resize the Vec with default values to offer a slice, but those default values are then overwritten by further consumption.

impl<T> From<IntoConsumer<T>> for Vec<T> {
    fn from(value: IntoConsumer<T>) -> Self {
        let (mut v, len) = (value.0, value.1);
        v.truncate(len);
        v
    }
}

impl<T> Debug for IntoConsumer<T>
where
    T: Debug,
{
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_tuple("IntoConsumer")
            .field(&&self.0[..self.1])
            .finish()
    }
}

impl<T> IntoConsumer<T> {
    /// Exposes all items consumed so far as a slice.
    ///
    /// ```
    /// use ufotofu::prelude::*;
    /// # pollster::block_on(async{
    /// let mut c = vec![].into_consumer();
    ///
    /// c.consume_item(1).await?;
    /// c.consume_item(2).await?;
    /// c.consume_item(4).await?;
    ///
    /// assert_eq!(c.as_slice(), [1, 2, 4].as_slice());
    /// # Result::<(), Infallible>::Ok(())
    /// # });
    /// ```
    pub fn as_slice(&self) -> &[T] {
        &self.0[..self.1]
    }

    /// Exposes all items consumed so far as a mutable slice.
    ///
    /// ```
    /// use ufotofu::prelude::*;
    /// # pollster::block_on(async{
    /// let mut c = vec![].into_consumer();
    ///
    /// c.consume_item(1).await?;
    /// c.consume_item(2).await?;
    /// c.consume_item(4).await?;
    ///
    /// assert_eq!(c.as_mut_slice(), [1, 2, 4].as_mut_slice());
    /// # Result::<(), Infallible>::Ok(())
    /// # });
    /// ```
    pub fn as_mut_slice(&mut self) -> &mut [T] {
        &mut self.0[..self.1]
    }

    /// Ensures that the next call to `self.expose_slots` will expose at least the requested number of slots.
    ///
    /// ```
    /// use ufotofu::prelude::*;
    /// # pollster::block_on(async{
    /// let mut c = Vec::<u32>::new().into_consumer();
    ///
    /// c.prepare_slots(17);
    /// c.expose_slots(async |slots| {
    ///     assert!(slots.len() >= 17);
    ///     (0, ())
    /// }).await?;
    /// # Result::<(), Infallible>::Ok(())
    /// # });
    /// ```
    pub fn prepare_slots(&mut self, amount: usize)
    where
        T: Default,
    {
        let old_len = self.0.len();
        self.0.resize_with(old_len + amount, Default::default);
    }
}

impl<T> Consumer for IntoConsumer<T> {
    type Item = T;
    type Final = ();
    type Error = Infallible;

    /// Appends the item to the vec.
    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        match val {
            Left(item) => {
                if self.0.len() == self.1 {
                    self.0.push(item);
                } else {
                    debug_assert!(self.0.len() > self.1);
                    self.0[self.1] = item;
                }

                self.1 += 1;
                Ok(())
            }
            Right(()) => Ok(()),
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        Ok(())
    }
}

impl<T: Default> BulkConsumer for IntoConsumer<T> {
    async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
    where
        F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
    {
        let len = self.0.len() - self.1;

        if len == 0 {
            let new_len = self.1 * 2 + 1;
            self.0.resize_with(new_len, Default::default);
        }

        let (amount, ret) = f(&mut self.0[self.1..]).await;
        self.1 += amount;
        Ok(ret)
    }
}

impl<T> crate::IntoConsumer for Vec<T> {
    type Item = T;
    type Final = ();
    type Error = Infallible;
    type IntoConsumer = IntoConsumer<T>;

    fn into_consumer(self) -> Self::IntoConsumer {
        let len = self.len();
        IntoConsumer(self, len)
    }
}

/// The consumer of the [`IntoConsumer`] impl of `&mut Vec<T>`; it appends consumed data to the [`Vec`].
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let mut v = vec![];
/// let mut c = (&mut v).into_consumer();
///
/// c.consume_item(1).await?;
/// c.consume_item(2).await?;
/// c.consume_item(4).await?;
///
/// drop(c);
///
/// assert_eq!(v, vec![1, 2, 4]);
/// # Result::<(), Infallible>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: the [producer::compat::vec::IntoProducerRef] type.
pub struct IntoConsumerMut<'a, T> {
    inner: &'a mut Vec<T>,
    items: usize,
    bulk_initialised: usize,
}

impl<'a, T> Debug for IntoConsumerMut<'a, T>
where
    T: Debug,
{
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_tuple("IntoConsumerMut")
            .field(&&self.inner[..self.items])
            .finish()
    }
}

impl<'a, T> Drop for IntoConsumerMut<'a, T> {
    fn drop(&mut self) {
        // Recover any hidden default items created by expose_slots_gracefully...
        unsafe {
            // Safety:
            // Whichever is larger of self.bulk_initialised and self.items is the number of items known to be initialised.
            self.inner.set_len(self.bulk_initialised.max(self.items));
        }

        // ...and drop them.
        self.inner.truncate(self.items);
    }
}

impl<'a, T> Consumer for IntoConsumerMut<'a, T> {
    type Item = T;
    type Final = ();
    type Error = Infallible;

    /// Appends the item to the vec.
    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        match val {
            Left(item) => {
                if self.inner.len() == self.items {
                    self.inner.push(item);
                } else {
                    debug_assert!(self.inner.len() > self.items);
                    self.inner[self.items] = item;
                }

                self.items += 1;
                Ok(())
            }
            Right(()) => Ok(()),
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        Ok(())
    }
}

impl<'a, T: Default> BulkConsumer for IntoConsumerMut<'a, T> {
    async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
    where
        F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
    {
        let len = self.inner.len() - self.items;
        let resize = len == 0;

        if resize {
            self.bulk_initialised = self.items * 2 + 1;
            self.inner
                .resize_with(self.bulk_initialised, Default::default);
        }

        let (amount, ret) = f(&mut self.inner[self.items..]).await;
        self.items += amount;

        // Avoid exposing default items in case the consumer is forgotten.
        if resize {
            // Safety:
            // self.items is necessarily <= the number of initialised items
            unsafe {
                self.inner.set_len(self.items);
            }
        }

        Ok(ret)
    }
}

impl<'a, T> crate::IntoConsumer for &'a mut Vec<T> {
    type Item = T;
    type Final = ();
    type Error = Infallible;
    type IntoConsumer = IntoConsumerMut<'a, T>;

    fn into_consumer(self) -> Self::IntoConsumer {
        let len = self.len();

        IntoConsumerMut {
            inner: self,
            items: len,
            bulk_initialised: len,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::prelude::*;
    use alloc::vec;
    use core::mem::forget;

    #[test]
    fn forget_ref_mut_matches_drop_ref_mut() {
        let mut v1 = vec![0, 1, 2];
        let mut v2 = v1.clone();

        let mut dropped = (&mut v1).into_consumer();
        let mut forgotten = (&mut v2).into_consumer();

        let new_items = vec![17; 17];

        pollster::block_on(async {
            assert_eq!(
                dropped.bulk_consume_full_slice(&new_items[..]).await,
                Ok(())
            );
            assert_eq!(
                forgotten.bulk_consume_full_slice(&new_items[..]).await,
                Ok(())
            );
        });

        drop(dropped);
        forget(forgotten);

        assert_eq!(v1, v2);
    }
}