ufotofu 0.12.5

//! Producer functionality for [`Vec`].
//!
//! Specifically, the module provides
//!
//! - an [`IntoProducer`] impl for `Vec<T>`,
//! - an [`IntoProducer`] impl for `&Vec<T>`, and
//! - an [`IntoProducer`] impl for `&mut Vec<T>`.
//!
//! <br/>Counterpart: the [`consumer::compat::vec`] module.

use core::{
    convert::Infallible,
    fmt,
    marker::PhantomData,
    ptr::{self, NonNull},
    slice,
};

#[cfg(feature = "alloc")]
use alloc::vec::Vec;

use crate::prelude::*;
use crate::producer::compat::{iterator_to_producer, IteratorToProducer};
use crate::producer::IntoProducer as _;

// `<Vec as IntoProducer>::IntoProducer` adapted from `<Vec as IntoIterator>::IntoIter`.

macro_rules! non_null {
    (mut $place:expr, $t:ident) => {{
        unsafe { &mut *((&raw mut $place) as *mut NonNull<$t>) }
    }};
    ($place:expr, $t:ident) => {{
        {
            #[allow(unused_unsafe)]
            unsafe {
                *((&raw const $place) as *const NonNull<$t>)
            }
        }
    }};
}

/// The producer of the [`IntoProducer`] impl of [`Vec`].
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let v = vec![1, 2, 4];
/// let mut p = v.into_producer();
///
/// assert_eq!(p.produce().await?, Left(1));
/// assert_eq!(p.produce().await?, Left(2));
/// assert_eq!(p.produce().await?, Left(4));
/// assert_eq!(p.produce().await?, Right(()));
/// # Result::<(), Infallible>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: [`consumer::compat::vec::IntoConsumer`].
pub struct IntoProducer<T> {
    buf: NonNull<T>,
    phantom: PhantomData<T>,
    cap: usize,
    ptr: NonNull<T>,
    /// If T is a ZST, this is actually ptr+len. This encoding is picked so that
    /// ptr == end is a quick test for the Iterator being empty, that works
    /// for both ZST and non-ZST.
    /// For non-ZSTs the pointer is treated as `NonNull<T>`
    end: *const T,
}

#[cfg(feature = "alloc")]
impl<T> crate::IntoProducer for Vec<T> {
    type Item = T;
    type Final = ();
    type Error = Infallible;
    type IntoProducer = IntoProducer<T>;

    fn into_producer(self) -> Self::IntoProducer {
        unsafe {
            let (data_ptr, len, cap) = self.into_raw_parts();
            let buf = NonNull::new_unchecked(data_ptr);
            let begin = buf.as_ptr();
            let end = if size_of::<T>() == 0 {
                begin.wrapping_byte_add(len)
            } else {
                begin.add(len) as *const T
            };
            IntoProducer {
                buf,
                phantom: PhantomData,
                cap,
                ptr: buf,
                end,
            }
        }
    }
}

impl<T: fmt::Debug> fmt::Debug for IntoProducer<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("IntoProducer")
            .field(&self.as_slice())
            .finish()
    }
}

impl<T> IntoProducer<T> {
    /// Returns the remaining items of this producer as a slice.
    pub fn as_slice(&self) -> &[T] {
        unsafe { slice::from_raw_parts(self.ptr.as_ptr(), self.len()) }
    }

    /// Returns the remaining items of this producer as a mutable slice.
    ///
    /// # Examples
    ///
    /// ```
    /// use ufotofu::prelude::*;
    /// let vec = vec!['a', 'b', 'c'];
    /// let mut producer = vec.into_producer();
    /// assert_eq!(producer.as_slice(), &['a', 'b', 'c']);
    /// producer.as_mut_slice()[2] = 'z';
    /// # pollster::block_on( async {
    /// assert_eq!(producer.produce().await.unwrap(), Left('a'));
    /// assert_eq!(producer.produce().await.unwrap(), Left('b'));
    /// assert_eq!(producer.produce().await.unwrap(), Left('z'));
    /// assert_eq!(producer.produce().await.unwrap(), Right(()));
    /// # });
    /// ```
    pub fn as_mut_slice(&mut self) -> &mut [T] {
        unsafe { &mut *self.as_raw_mut_slice() }
    }

    fn as_raw_mut_slice(&mut self) -> *mut [T] {
        ptr::slice_from_raw_parts_mut(self.ptr.as_ptr(), self.len())
    }

    #[inline]
    fn len(&self) -> usize {
        if size_of::<T>() == 0 {
            self.end.addr().wrapping_sub(self.ptr.as_ptr().addr())
        } else {
            unsafe { non_null!(self.end, T).offset_from_unsigned(self.ptr) }
        }
    }
}

impl<T> AsRef<[T]> for IntoProducer<T> {
    fn as_ref(&self) -> &[T] {
        self.as_slice()
    }
}

impl<T> Producer for IntoProducer<T> {
    type Item = T;

    type Final = ();

    type Error = Infallible;

    async fn produce(&mut self) -> Result<Either<Self::Item, Self::Final>, Self::Error> {
        let ptr = if size_of::<T>() == 0 {
            if core::ptr::eq(self.ptr.as_ptr(), self.end) {
                return Ok(Right(()));
            }
            // `ptr` has to stay where it is to remain aligned, so we reduce the length by 1 by
            // reducing the `end`.
            self.end = self.end.wrapping_byte_sub(1);
            self.ptr
        } else {
            if self.ptr == non_null!(self.end, T) {
                return Ok(Right(()));
            }
            let old = self.ptr;
            self.ptr = unsafe { old.add(1) };
            old
        };
        Ok(Left(unsafe { ptr.read() }))
    }

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

#[cfg(feature = "alloc")]
impl<T> BulkProducer for IntoProducer<T> {
    async fn expose_items_gracefully<F, R>(
        &mut self,
        f: F,
    ) -> Result<Either<R, (F, Self::Final)>, (F, Self::Error)>
    where
        F: AsyncFnOnce(&[Self::Item]) -> (usize, R),
    {
        if self.len() == 0 {
            Ok(Right((f, ())))
        } else {
            let (produced, result) = (f)(self.as_slice()).await;
            assert!(
                produced <= self.len(),
                "f must not claim to have processed more items than present in the passed slice"
            );
            if size_of::<T>() == 0 {
                self.end = self.end.wrapping_byte_sub(produced)
            } else {
                self.ptr = unsafe { self.ptr.add(produced) };
            }
            Ok(Left(result))
        }
    }
}

#[cfg(feature = "alloc")]
impl<T: Clone> Clone for IntoProducer<T> {
    fn clone(&self) -> Self {
        self.as_slice().to_vec().into_producer()
    }
}

#[cfg(feature = "alloc")]
impl<T> Default for IntoProducer<T> {
    fn default() -> Self {
        Vec::default().into_producer()
    }
}

#[cfg(feature = "alloc")]
impl<T> Drop for IntoProducer<T> {
    fn drop(&mut self) {
        struct DropGuard<'a, T>(&'a mut IntoProducer<T>);

        impl<T> Drop for DropGuard<'_, T> {
            fn drop(&mut self) {
                unsafe {
                    // Vec handles deallocation
                    let _ = Vec::from_raw_parts(self.0.buf.as_ptr(), 0, self.0.cap);
                }
            }
        }

        let guard = DropGuard(self);
        // destroy the remaining elements
        unsafe {
            ptr::drop_in_place(guard.0.as_raw_mut_slice());
        }
        // now `guard` will be dropped and do the rest
    }
}

/// The producer of the [`IntoProducer`] impl of `&Vec<T>`.
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let vec_ref = &vec![1, 2, 4];
/// let mut p = vec_ref.into_producer();
///
/// assert_eq!(p.produce().await?, Left(&1));
/// assert_eq!(p.produce().await?, Left(&2));
/// assert_eq!(p.produce().await?, Left(&4));
/// assert_eq!(p.produce().await?, Right(()));
/// # Result::<(), Infallible>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: [`consumer::compat::vec::IntoConsumerMut`].
#[cfg(feature = "alloc")]
#[derive(Debug, Clone)]
pub struct IntoProducerRef<'s, T>(IteratorToProducer<<&'s Vec<T> as IntoIterator>::IntoIter>);

#[cfg(feature = "alloc")]
impl<'s, T> Producer for IntoProducerRef<'s, T> {
    type Item = &'s T;
    type Final = ();
    type Error = Infallible;

    async fn produce(&mut self) -> Result<Either<Self::Item, Self::Final>, Self::Error> {
        self.0.produce().await
    }

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

#[cfg(feature = "alloc")]
impl<'s, T> crate::IntoProducer for &'s Vec<T> {
    type Item = &'s T;
    type Final = ();
    type Error = Infallible;
    type IntoProducer = IntoProducerRef<'s, T>;

    fn into_producer(self) -> Self::IntoProducer {
        IntoProducerRef(iterator_to_producer(self.iter()))
    }
}

/// The producer of the [`IntoProducer`] impl of `&mut Vec<T>`.
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let mut v = vec![1, 2, 4];
/// let mut p = (&mut v).into_producer();
///
/// assert_eq!(p.produce().await?, Left(&mut 1));
/// let mut mutable_ref = p.produce().await?.unwrap_left();
/// *mutable_ref = 17;
/// assert_eq!(p.produce().await?, Left(&mut 4));
/// assert_eq!(p.produce().await?, Right(()));
/// assert_eq!(v, vec![1, 17, 4]);
/// # Result::<(), Infallible>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: none, because you cannot consume into a collection behind an immutable reference.
#[cfg(feature = "alloc")]
pub struct IntoProducerMut<'s, T>(IteratorToProducer<<&'s mut Vec<T> as IntoIterator>::IntoIter>);

#[cfg(feature = "alloc")]
impl<'s, T> Producer for IntoProducerMut<'s, T> {
    type Item = &'s mut T;
    type Final = ();
    type Error = Infallible;

    async fn produce(&mut self) -> Result<Either<Self::Item, Self::Final>, Self::Error> {
        self.0.produce().await
    }

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

#[cfg(feature = "alloc")]
impl<'s, T> crate::IntoProducer for &'s mut Vec<T> {
    type Item = &'s mut T;
    type Final = ();
    type Error = Infallible;
    type IntoProducer = IntoProducerMut<'s, T>;

    fn into_producer(self) -> Self::IntoProducer {
        IntoProducerMut(iterator_to_producer(self.iter_mut()))
    }
}

#[cfg(test)]
mod tests {

    use crate::prelude::*;
    use crate::ProduceAtLeastError;
    use alloc::vec;

    #[test]
    fn vec_bulk_producer_slice_overwrite() {
        let mut s = [0; 17];

        let mut p1 = vec![17; 17].into_producer();
        pollster::block_on(async {
            assert_eq!(p1.bulk_overwrite_full_slice(&mut s).await, Ok(()))
        });
        assert_eq!(s, [17; 17]);

        let mut p2 = vec![0; 1024].into_producer();
        pollster::block_on(async {
            assert_eq!(p2.bulk_overwrite_full_slice(&mut s).await, Ok(()))
        });
        assert_eq!(s, [0; 17]);

        let mut p3 = vec![17; 5].into_producer();
        pollster::block_on(async {
            assert_eq!(
                p3.bulk_overwrite_full_slice(&mut s).await,
                Err(ProduceAtLeastError {
                    count: 5,
                    reason: Ok(())
                })
            );
        });
        assert_eq!(s[..5], [17; 5]);
        assert_eq!(s[5..], [0; 12]);

        assert_eq!(p1.as_slice().len(), 0);
        assert_eq!(p2.as_slice().len(), 1007);
        assert_eq!(p3.as_slice().len(), 0);
    }

    #[test]
    fn vec_bulk_produce_zst() {
        let mut s = [(); 17];

        let mut p1 = vec![(); 17].into_producer();
        pollster::block_on(async {
            assert_eq!(p1.bulk_overwrite_full_slice(&mut s).await, Ok(()))
        });

        let mut p2 = vec![(); 1024].into_producer();
        pollster::block_on(async {
            assert_eq!(p2.bulk_overwrite_full_slice(&mut s).await, Ok(()))
        });

        let mut p3 = vec![(); 5].into_producer();
        pollster::block_on(async {
            assert_eq!(
                p3.bulk_overwrite_full_slice(&mut s).await,
                Err(ProduceAtLeastError {
                    count: 5,
                    reason: Ok(())
                })
            );
        });

        assert_eq!(p1.as_slice().len(), 0);
        assert_eq!(p2.as_slice().len(), 1007);
        assert_eq!(p3.as_slice().len(), 0);
    }
}