//! Consumers — values that asynchronously process a sequence of items.
//!
//! A [`Consumer`] processed items of type [`Consumer::Item`], fed to it as [`Left`] arguments to the [`Consumer::consume`] method. After the calling code has moved all items into the consumer, it may call [`Consumer::consume`] with a [`Right`] argument of type [`Consumer::Final`] to signal to the consumer that no more items will follow.
//!
//! ```
//! use ufotofu::prelude::*;
//! # #[cfg(feature = "alloc")] {
//! # pollster::block_on(async{
//! let mut my_first_consumer = vec![].into_consumer();
//!
//! my_first_consumer.consume(Left(1)).await?;
//! my_first_consumer.consume(Left(2)).await?;
//! my_first_consumer.consume(Left(4)).await?;
//!
//! let vec: Vec<_> = my_first_consumer.into();
//! assert_eq!(vec, vec![1, 2, 4]);
//! # Result::<(), Infallible>::Ok(())
//! # });
//! # }
//! ```
//!
//! Consumers may emit errors of type [`Consumer::Error`] to indicate failure to process a regular item or the final value. It is forbidden to call `consume` after having called it with a final value, or after the consumer has emitted an error. Any such call may result in unspecified (but safe) behaviour.
//!
//! <br/>
//!
//! The [`IntoConsumer`] trait describes types which can be converted into consumers. This trait is implemented for the collection types of the standard library. It is also implemented on *mutable references* to such collections, allowing you to consume into a collection without taking ownership of it:
//!
//! ```
//! use ufotofu::prelude::*;
//! # #[cfg(feature = "alloc")] {
//! # pollster::block_on(async{
//! let mut v = vec![];
//! let mut c = (&mut v).into_consumer();
//!
//! c.consume(Left(1)).await?;
//! c.consume(Left(2)).await?;
//! c.consume(Left(4)).await?;
//!
//! drop(c);
//!
//! assert_eq!(v, vec![1, 2, 4]);
//! # Result::<(), Infallible>::Ok(())
//! # });
//! # }
//! ```
//!
//! <br/>
//!
//! Every consumer may delay performing side-effects to make `consume` calls more efficient. The classic example of a buffering consumer is a consumer of bytes which writes to a file from disk: it most certainly should not write every individual byte to disk, instead it should buffer bytes in memory and occasionally flush the buffer to disk.
//!
//! The [`flush`](Consumer::flush) method lets calling code instruct the consumer to immediately perform the observable side-effects for all currently buffered data.
//!
//! <br/>
//!
//! Every consumer automatically implements the [`ConsumerExt`] trait, which provides useful methods for working with consumers. In particular, it provides the [`consume_item`](ConsumerExt::consume_item) and [`consume_final`](ConsumerExt::consume_final) methods for conveniently consuming values without wrapping them in [`Eithers`](Either):
//!
//! ```
//! use ufotofu::prelude::*;
//! # #[cfg(feature = "alloc")] {
//! # 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/>
//!
//! Consuming a sequence one item at a time can be inefficient. The [`BulkConsumer`] trait extends [`Consumer`] with the ability to consume multiple items at a time. This is enabled by the [`BulkConsumer::expose_slots_gracefully`] method. You pass to this method an async function as the sole argument. The bulk consumer calls that function, passing it a mutable, non-empty slice of items. The function can overwrite items in that buffer, and then returns a pair of values: first, the number of items from the buffer the consumer should now consume, and second, an arbitrary value, to be returned by the `expose_slots_gracefully` call.
//! If the consumer reports an error when `expose_slots_gracefully` is called, the provided function is returned alongside the error value.
//! ```
//! use ufotofu::prelude::*;
//! # pollster::block_on(async{
//! let mut arr = [0, 0, 0];
//! let mut c = (&mut arr).into_consumer();
//!
//! assert_eq!(c.expose_slots_gracefully(async |mut slots| {
//! slots[0] = 1;
//! slots[1] = 2;
//! slots[2] = 4;
//! (3, "hi!")
//! }).await.map_err(|_| ())?, "hi!");
//! assert_eq!(c.consume_item(8).await, Err(()));
//!
//! assert_eq!(arr, [1, 2, 4]);
//!
//! // If we reported that we only wrote two items, the consumer would later accept another item:
//! let mut arr2 = [0, 0, 0];
//! let mut c2 = (&mut arr2).into_consumer();
//!
//! assert_eq!(c2.expose_slots_gracefully(async |mut slots| {
//! slots[0] = 1;
//! slots[1] = 2;
//! slots[2] = 4;
//! (2, "hi!")
//! }).await.map_err(|_| ())?, "hi!");
//! assert_eq!(c2.consume_item(8).await, Ok(()));
//!
//! assert_eq!(arr2, [1, 2, 8]);
//! # Result::<(), ()>::Ok(())
//! # });
//! ```
//!
//! <br/>
//!
//! Every bulk consumer automatically implements the [`BulkConsumerExt`] trait, which provides bulk-consumption-based variants of several methods of [`ConsumerExt`]. These bulk versions are typically more efficient and should be preferred whenever possible.
//!
//! Of particular note are the following:
//! - The [`BulkConsumerExt::expose_slots`] method, which behaves exactly like `expose_slots_gracefully` except that the provided function is not returned alongside error values to allow for simplified error handling.
//! - The [`BulkConsumerExt::bulk_consume`] method, which builds on `expose_slots` and reimplements the way that, e.g., [`std::io::Write`] accepts multiple items at a time: `bulk_consume` takes a slice as its input, and the consumer reports how many items from that slice it consumed.
//!
//! ```
//! use ufotofu::prelude::*;
//! # pollster::block_on(async{
//! let mut arr = [0, 0, 0];
//!
//! let mut c1 = (&mut arr).into_consumer();
//!
//! assert_eq!(c1.expose_slots(async |slots| {
//! slots[0] = 0;
//! slots[1] = 1;
//! slots[2] = 2;
//! (3, "hi!")
//! }).await?, "hi!");
//! assert_eq!(arr, [0, 1, 2]);
//!
//! let mut c2 = (&mut arr).into_consumer();
//!
//! assert_eq!(c2.bulk_consume(&[1, 2]).await?, 2);
//! assert_eq!(c2.bulk_consume(&[4, 8]).await?, 1);
//!
//! assert_eq!(arr, [1, 2, 4]);
//! # Result::<(), ()>::Ok(())
//! # });
//! ```
//!
//! <br/>The counterpart to the [`consumer`] module is the [`producer`] module.
use crate::prelude::*;
mod consumer_ext;
pub use consumer_ext::*;
mod move_into_slice;
pub use move_into_slice::*;
mod full;
pub use full::*;
mod error_immediately;
pub use error_immediately::*;
pub mod compat;
mod buffered;
pub use buffered::*;
mod bulk_buffered;
pub use bulk_buffered::*;
mod map_err;
pub use map_err::*;
mod map_final;
pub use map_final::*;
mod map_item;
pub use map_item::*;
mod chain;
pub use chain::*;
mod consume_while;
pub use consume_while::*;
mod limit;
pub use limit::*;
#[cfg(feature = "alloc")]
mod stats;
#[cfg(feature = "alloc")]
pub use stats::*;
mod filter;
pub use filter::*;
#[cfg(feature = "alloc")]
mod watch;
#[cfg(feature = "alloc")]
pub use watch::*;
#[cfg(feature = "dev")]
mod scrambled;
#[cfg(feature = "dev")]
pub use scrambled::*;
#[cfg(feature = "dev")]
mod bulk_scrambled;
#[cfg(feature = "dev")]
pub use bulk_scrambled::*;
#[cfg(feature = "dev")]
mod test_consumer;
#[cfg(feature = "dev")]
pub use test_consumer::*;
/// A [`Consumer`] lazily processes a sequence of items.
///
/// The sequence consists of an arbitrary number of items of type [`Consumer::Item`], and is optionally terminated by a value of type [`Consumer::Final`]. At any point, a consumer may report an error of type [`Consumer::Error`].
///
/// ```
/// use ufotofu::prelude::*;
/// # #[cfg(feature = "alloc")] {
/// # pollster::block_on(async{
/// let mut v = vec![];
/// let mut c = (&mut v).into_consumer();
///
/// c.consume(Left(1)).await?;
/// c.consume(Left(2)).await?;
/// c.consume(Left(4)).await?;
///
/// drop(c);
///
/// assert_eq!(v, vec![1, 2, 4]);
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// Every consumer may delay performing side-effects to make `consume` calls more efficient. The classic example of a buffering consumer is a consumer of bytes which writes to a file from disk: it most certainly should not write every individual byte to disk, instead it should buffer bytes in memory and occasionally flush the buffer to disk.
///
/// The [`flush`](Consumer::flush) method lets calling code instruct the consumer to immediately perform the observable side-effects for all currently buffered data.
///
/// Calling code must uphold the following invariants:
///
/// - Do not call trait methods after closing the consumer, i.e., after calling [`consume`](Consumer::consume) with a [`Right`] value.
/// - Do not call trait methods after any method has yielded an error.
/// - Do not use the consumer after dropping any `Future` returned by any of its methods, unless the dropped future has been polled to completion.
/// - Do not use the consumer after catching an unwinding panic.
///
/// <br/>Counterpart: the [`Producer`] trait.
#[must_use = "consumers are lazy and do nothing unless fed with values and/or closed"]
pub trait Consumer {
/// The sequence processed by this consumer starts with *arbitrarily many* values of this type.
type Item;
/// The sequence processed by this consumer ends with *up to one* value of this type.
type Final;
/// The type of errors the consumer can emit instead of doing its job.
type Error;
/// Attempts to process the next sequence value. The method may fail, returning an `Err` instead.
///
/// We say the conumer "has been closed" after this method has been called with a `Right` final value.
///
/// After this method returns an error, no further methods of this trait may be invoked.
///
/// # Invariants
///
/// Must not be called after this consumer has been closed, or after any method of this trait has returned an error.
///
/// <br/>Counterpart: the [`Producer::produce`] method.
async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error>;
/// Attempts to perform any effectful actions that were delayed to make preceding calls to `consume` more efficient.
///
/// This function allows calling code to trigger side-effects, which otherwise can only be triggered deliberately by closing the consumer, i.e. by calling [`Consumer::consume`] with a `Right` final value.
///
/// After this function returns an error, no further methods of this trait may be invoked.
///
/// #### Invariants
///
/// Must not be called after this consumer has been closed, or after any method of this trait has returned an error.
///
/// <br/>Counterpart: the [`Producer::slurp`] method.
async fn flush(&mut self) -> Result<(), Self::Error>;
}
impl<C> Consumer for &mut C
where
C: Consumer + ?Sized,
{
type Item = C::Item;
type Final = C::Final;
type Error = C::Error;
async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
(*self).consume(val).await
}
async fn flush(&mut self) -> Result<(), Self::Error> {
(*self).flush().await
}
}
#[cfg(feature = "alloc")]
impl<C> Consumer for alloc::boxed::Box<C>
where
C: Consumer + ?Sized,
{
type Item = C::Item;
type Final = C::Final;
type Error = C::Error;
async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
self.as_mut().consume(val).await
}
async fn flush(&mut self) -> Result<(), Self::Error> {
self.as_mut().flush().await
}
}
impl Consumer for Infallible {
type Item = Infallible;
type Final = Infallible;
type Error = Infallible;
async fn consume(&mut self, _val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
unreachable!()
}
async fn flush(&mut self) -> Result<(), Self::Error> {
unreachable!()
}
}
/// Conversion into a [`Consumer`].
///
/// By implementing `IntoConsumer` for a type, you define how it will be
/// converted to a consumer. This is common for types which describe a
/// collection of some kind.
///
/// <br/>Counterpart: the [`IntoProducer`] trait.
pub trait IntoConsumer {
/// The type of repeated items being consumed.
type Item;
/// The type of the final value being consumed.
type Final;
/// The type of errors the consumer may emit.
type Error;
/// The consumer into which values of this type can be converted.
///
/// <br/>Counterpart: the [`IntoProducer::IntoProducer`] type.
type IntoConsumer: Consumer<Item = Self::Item, Final = Self::Final, Error = Self::Error>;
/// Creates a consumer from a value.
///
/// <br/>Counterpart: the [`IntoProducer::into_producer`] method.
fn into_consumer(self) -> Self::IntoConsumer;
}
impl<C: Consumer> IntoConsumer for C {
type Item = C::Item;
type Final = C::Final;
type Error = C::Error;
type IntoConsumer = C;
#[inline]
fn into_consumer(self) -> C {
self
}
}
impl IntoConsumer for () {
type Item = Infallible;
type Final = ();
type Error = Infallible;
type IntoConsumer = Full<()>;
#[inline]
fn into_consumer(self) -> Self::IntoConsumer {
full()
}
}
/// A [`BulkConsumer`] is a producer that can accept multiple items with a single call of the [`BulkConsumer::expose_slots_gracefully`] method.
///
/// This method takes an async function as its sole argument. The consumer calls that function, passing it a mutable, non-empty slice of items. The function can mutate these items in any way, and then returns a pair of values: first, the number of items the consumer should now consider as having been consumed, and second, an arbitrary value, to be returned by the `expose_slots` call.
///
/// See [`BulkConsumerExt::expose_slots`] for a variant of `expose_slots_gracefully` which simplifies error handling by not returning the supplied function.
/// See [`BulkConsumerExt::bulk_consume`] for using bulk consumers in a way analogous to [`std::io::Write::write`].
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let mut arr = [0, 0, 0];
/// let mut c = (&mut arr).into_consumer();
///
/// assert_eq!(c.expose_slots(async |mut slots| {
/// slots[0] = 1;
/// slots[1] = 2;
/// slots[2] = 4;
/// (3, "hi!")
/// }).await?, "hi!");
/// assert_eq!(c.consume_item(8).await, Err(()));
///
/// assert_eq!(arr, [1, 2, 4]);
///
/// // If we reported that we only wrote two items, the consumer would later accept another item:
/// let mut arr2 = [0, 0, 0];
/// let mut c2 = (&mut arr2).into_consumer();
///
/// assert_eq!(c2.expose_slots(async |mut slots| {
/// slots[0] = 1;
/// slots[1] = 2;
/// slots[2] = 4;
/// (2, "hi!")
/// }).await?, "hi!");
/// assert_eq!(c2.consume_item(8).await, Ok(()));
///
/// assert_eq!(arr2, [1, 2, 8]);
/// # Result::<(), ()>::Ok(())
/// # });
/// ```
///
/// Semantically, there should be no difference between bulk consumption or item-by-item consumption.
///
/// <br/>Counterpart: the [`BulkProducer`] trait.
pub trait BulkConsumer: Consumer {
/// Exposes a non-empty number of item slots to a given async function, the function mutates those slots and then then reports to the consumer how many items should now be considered consumed.
///
/// When the consumer needs to return an error, it must return it immediately from this method. Otherwise, i.e., when it wants to consume regular items, it must call the passed function `f` with a non-empty slice of items as the argument, and then poll `f` to completion. When `f` has yielded `(amount, t)`, the consumer must adjust its state as if `consume` had been called `amount` many times with copies of the first `amount` many items in the slice it had passed to `f`, and then return `Ok(t)`. It must not return `Err(_)` when having called `f`.
///
/// When passing a slice to `f`, if `f` does not report that *every* provided slot was filled, then the consumer must be able to consume the remaining number of items without emitting an error. For example: a consumer passes a slice of five slots to `f`, and `f` returns `(3, _)`. Then the next two calls to `consume` *must* return `Ok(())`.
///
/// After this function returns the final value or after it returns an error, no further methods of this trait may be invoked.
///
/// The intention is for `f` to not read any contents of the passed slice, but to simply write items into the slice that should be consumed. Sadly, we cannot enforce this on the type level.
///
/// Because `f` might still be needed after the consumer reports an error, this method returns `f` alongside its error value. For more convenient error handling with the `?` operator in cases where `f` does not need to be preserved, see [`expose_slots`](`BulkConsumerExt::expose_slots`).
///
/// # Invariants
///
/// Must not be called after this consumer has been closed, or after any method of this trait has returned an error.
///
/// `f` must not return an `amount` strictly greater than the length of the buffer passed to it.
///
/// # Examples
///
/// ```
/// use ufotofu::prelude::*;
/// # pollster::block_on(async{
/// let mut arr = [0, 0, 0];
/// let mut c = (&mut arr).into_consumer();
///
/// assert_eq!(c.expose_slots_gracefully(async |mut slots| {
/// slots[0] = 1;
/// slots[1] = 2;
/// slots[2] = 4;
/// (3, "hi!")
/// }).await.map_err(|(_f, err)| err), Ok("hi!"));
/// assert_eq!(c.consume_item(8).await, Err(()));
///
/// assert_eq!(arr, [1, 2, 4]);
///
/// // If we reported that we only wrote two items, the consumer would later accept another item:
/// let mut arr2 = [0, 0, 0];
/// let mut c2 = (&mut arr2).into_consumer();
///
/// assert_eq!(c2.expose_slots_gracefully(async |mut slots| {
/// slots[0] = 1;
/// slots[1] = 2;
/// slots[2] = 4;
/// (2, "hi!")
/// }).await.map_err(|(_f, err)| err), Ok("hi!"));
/// assert_eq!(c2.consume_item(8).await, Ok(()));
///
/// assert_eq!(arr2, [1, 2, 8]);
///
/// // If the consumer reports an error on a call to `expose_slots_gracefully`, the supplied async function is returned for later use:
///
/// let mut no_room: [u8; 0] = [];
/// let mut overflow = [0u8; 3];
/// let mut c3 = (&mut no_room).into_consumer();
/// let mut c4 = (&mut overflow).into_consumer();
///
/// let f = match c3.expose_slots_gracefully(async |mut slots| {
/// slots[0] = 17;
/// (1, "hi!")
/// }).await {
/// Err((f, _err)) => f,
/// Ok(_) => unreachable!(),
/// };
///
/// assert_eq!(c4.expose_slots_gracefully(f).await.map_err(|(_f, err)| err), Ok("hi!"));
/// assert_eq!(overflow, [17, 0, 0]);
///
/// # Result::<(), ()>::Ok(())
/// # });
/// ```
///
/// <br/>Counterpart: the [`BulkProducer::expose_items_gracefully`] method.
async fn expose_slots_gracefully<F, R>(&mut self, f: F) -> Result<R, (F, Self::Error)>
where
F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R);
}
impl<C> BulkConsumer for &mut C
where
C: BulkConsumer + ?Sized,
{
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).expose_slots_gracefully(f).await
}
}
#[cfg(feature = "alloc")]
impl<C> BulkConsumer for alloc::boxed::Box<C>
where
C: BulkConsumer + ?Sized,
{
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.as_mut().expose_slots_gracefully(f).await
}
}
impl BulkConsumer for Infallible {
async fn expose_slots_gracefully<F, R>(&mut self, _f: F) -> Result<R, (F, Self::Error)>
where
F: AsyncFnOnce(&mut [Self::Item]) -> (usize, R),
{
unreachable!()
}
}
/// Conversion into a [`BulkConsumer`].
///
/// This trait is automatically implemented by implementing [`IntoConsumer`] with the associated consumer being a bulk consumer.
///
/// <br/>Counterpart: the [`IntoBulkProducer`] trait.
pub trait IntoBulkConsumer: IntoConsumer<IntoConsumer: BulkConsumer> {}
impl<C> IntoBulkConsumer for C where C: IntoConsumer<IntoConsumer: BulkConsumer> {}